GGraziadei commented on code in PR #8950: URL: https://github.com/apache/storm/pull/8950#discussion_r3789325747
########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergCommitter.java: ########## @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.common; + +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Makes durable data files visible in an Iceberg table, atomically and recoverably. + * + * <p>A commit is prepared in the {@link CommitWal} first, then appended in a single Iceberg + * operation that stamps the commit id on the resulting snapshot, then cleared from the WAL. + * Because the append is atomic, readers never observe part of a batch. Because the snapshot + * carries the commit id, {@link #recover()} can tell a commit that landed from one that did not, + * without needing any identity from the source: the table itself answers the question. + * + * <p>This yields atomic commits with at-least-once delivery. A crash before the WAL entry exists + * leaves orphan data files, which are invisible to readers and removed by Iceberg's standard + * orphan-file maintenance; a replayed batch is written and committed again, and its rows stay + * visible until something downstream removes them. + * + * <p>One commit may cover many batches — the aggregated committer hands it the files of every + * writer it collected — but it is still a single atomic append carrying a single commit id. + */ +public class IcebergCommitter { + + public static final String COMMIT_ID_PROPERTY = "storm.iceberg.commit-id"; + /** + * How far before a WAL entry's own timestamp the snapshot scan still looks. The snapshot is + * always written after the entry, so only clock skew between the worker that wrote the entry + * and whatever stamped the snapshot's timestamp can put it earlier. + */ + static final long CLOCK_SKEW_ALLOWANCE_MS = TimeUnit.MINUTES.toMillis(10); + + private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class); + + private final Table table; + private final CommitWal wal; + private final IcebergMetrics metrics; + + public IcebergCommitter(Table table, CommitWal wal, IcebergMetrics metrics) { + this.table = table; + this.wal = wal; + this.metrics = metrics; + } + + /** + * Log, append and clear one commit. Does nothing when there is nothing to append. + * + * <p>Returns normally only when the batch is visible in the table — including the case where + * the append reported a failure but had in fact landed. Throwing means the batch is not + * visible and its tuples must be replayed. + */ + public void commit(List<DataFile> dataFiles) { + if (dataFiles.isEmpty()) { + return; + } + CommitWal.WalEntry entry = wal.write(dataFiles); + long startNanos = System.nanoTime(); + try { + append(entry, dataFiles); + } catch (RuntimeException e) { + settleFailedCommit(entry, dataFiles, startNanos, e); + return; + } + metrics.committed(dataFiles, System.nanoTime() - startNanos); + wal.delete(entry); + } + + /** + * Resolve a failed commit while the batch is still in hand, rather than leaving it to the next + * startup. Asking the table whether the commit landed turns an unknown outcome into a known + * one at the only moment when it can still be acted on. + * + * <p>If it landed, the batch is visible and the caller may ack. If it did not, the entry is + * dropped before the exception propagates: the caller will fail those tuples, the source will + * replay them, and a WAL entry left behind would make the next startup append the original + * files too — duplicating what the replay writes. The abandoned files become orphans instead, + * which is what orphan-file maintenance is for. + */ + private void settleFailedCommit(CommitWal.WalEntry entry, List<DataFile> dataFiles, + long startNanos, RuntimeException failure) { + boolean landed; + try { + landed = isVisible(entry); + } catch (RuntimeException e) { + // The table cannot be reached, so the outcome stays unknown. Leave the entry: startup + // will settle it, and replaying a commit is recoverable in a way that losing it is not. + metrics.commitFailed(); + failure.addSuppressed(e); + throw failure; + } + wal.delete(entry); + if (landed) { + // The data is visible, so it counts as committed however the append reported itself. + metrics.committed(dataFiles, System.nanoTime() - startNanos); + LOG.warn("Commit {} reported a failure but its snapshot is present; " + + "treating it as successful", entry.commitId(), failure); + return; + } + metrics.commitFailed(); + LOG.error("Commit {} did not land; its data files are left as orphans and its tuples " + + "will be replayed", entry.commitId(), failure); + throw failure; + } Review Comment: Confirmed, and not closable — a bounded re-check narrows the window without eliminating it, so the sink does not wait: it takes one sample and treats the outcome as final. What was wrong here was the documentation, which claimed otherwise. `docs/storm-iceberg.md` now states the mechanism you describe: a REST catalog maps 500/502/503/504 to `CommitStateUnknownException`, `SnapshotProducer.commit()` is `onlyRetryOn(CommitFailedException.class)`, so a commit the backend applies after the sample reads as absent and the replay writes those rows a second time. Duplicates are inside the contract this module offers, and that is now what the docs say rather than the opposite. ########## docs/storm-iceberg.md: ########## @@ -0,0 +1,322 @@ +--- +title: Storm Apache Iceberg Integration +layout: documentation +documentation: true +--- + +Bolt for writing data to [Apache Iceberg](https://iceberg.apache.org/) tables directly from a +Storm topology — no Kafka Connect or Spark job in between — with **atomic commits and +at-least-once delivery**. + +## Guarantees + +Read this before anything else; it is the part that decides whether this module fits. + +- **Atomic commits.** A batch becomes visible in one Iceberg append, or not at all. Readers never + see part of a batch, and a crash costs orphan data files rather than a broken table. +- **At-least-once delivery.** Tuples are acked only after the commit containing them has landed, + so nothing is silently lost. A batch that fails is replayed by the source and written again. +- **Duplicates are possible and are not removed.** The sink appends; it writes no equality + deletes. Rows from a replayed batch stay visible until something downstream deduplicates them. + +This module does **not** promise exactly-once. Exactly-once would need a deterministic identity of +the input — the same batch content under the same identifier on replay — and that comes from the +source, not from the sink. A general-purpose module cannot assume every user has a replayable, +deterministically addressed source, so the claim is not made. An extractor SPI for sources that +can do better is a possible future addition, not a current guarantee. + +The target must be an **append-only, format-version-2** Iceberg table. Row-level deletes, +upserts, and merge-on-read are out of scope. + +## Usage + +```java +Map<String, String> catalogProps = new HashMap<>(); +catalogProps.put("type", "rest"); +catalogProps.put("uri", "http://rest-catalog:8181";); + +IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events") + .withCommitIntervalBytes(128L * 1024 * 1024) + .build(); + +TopologyBuilder builder = new TopologyBuilder(); +builder.setSpout("events", spout, 2); +builder.setBolt("iceberg", new IcebergBolt(options), 4) + .shuffleGrouping("events"); + +Config conf = new Config(); +// Bounds how long a partial batch waits when the stream goes quiet. +conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, 30); +``` + +The catalog properties are passed verbatim to Iceberg's `CatalogUtil.buildIcebergCatalog(...)`, +so every Iceberg catalog works with its standard configuration keys: `type` = `hive`, `hadoop`, +`rest`, or `catalog-impl` for Glue, Nessie, JDBC, etc. The catalog implementations themselves are +**not** pulled in transitively — see [Dependencies](#dependencies). + +### Options + +| Option | Default | Description | +|---|---|---| +| `withCatalogProperties(Map)` | required | Iceberg catalog configuration | +| `withTable(String)` | required | Target table identifier, e.g. `db.events` | +| `withRecordMapper(RecordMapper)` | `FieldNameRecordMapper` | Tuple → `Record` conversion | +| `withFileFormat(FileFormat)` | `PARQUET` | Data file format | +| `withTargetFileSizeBytes(long)` | table property `write.target-file-size-bytes` | Rolling file size | +| `withAutoCreate(Schema, PartitionSpec)` | disabled | Create the table on first use if missing | +| `withCommitIntervalRecords(int)` | 1000, when no other threshold is set | Close the batch after this many tuples | +| `withCommitIntervalBytes(long)` | disabled | Close the batch after roughly this many bytes | +| `withCommitIntervalMillis(long)` | disabled | Close the batch once it has been open this long | +| `withGroupCommitIntervalMillis(long)` | 5000 | `IcebergCommitterBolt` only: commit once the oldest accumulated batch is this old | +| `withGroupCommitMaxDataFiles(int)` | 1000 | `IcebergCommitterBolt` only: commit once this many data files have accumulated | +| `withTickIntervalSecs(int)` | none — inherits `topology.tick.tuple.freq.secs` | Per-component tick frequency for the writer or committer | + +### Batch sizing + +Committing every tuple would mean one snapshot and one small file per tuple, which degrades +catalog and reader planning quickly. The bolt therefore accumulates tuples and commits them +together, closing the batch when the first configured threshold is crossed. If you configure none, +it falls back to `withCommitIntervalRecords(1000)` so a batch can never wait indefinitely. + +**Buffering costs latency and replay volume, not durability.** Buffered tuples are not acked, so a +worker that dies mid-batch has them replayed rather than losing them. The trade-off to weigh is +how much work a crash repeats, and how long rows wait before becoming visible — not whether they +survive. + +Configure `Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS` as well. `withCommitIntervalMillis` is evaluated +when the next tuple arrives, so on a stream that stops entirely only a tick tuple can close the +final partial batch. + +Set `topology.max.spout.pending` comfortably above the number of tuples one batch accumulates: a +batch's tuples stay un-acked until it commits, so too low a value stalls the topology. + +### Tuple mapping + +By default tuple fields are matched to table columns **by name** (`FieldNameRecordMapper`): +numeric values are widened to the column type, `Instant` / `java.util.Date` / epoch-millis +`Long` values are converted for timestamp columns, `byte[]` becomes `ByteBuffer`. A required +column with no tuple value fails the topology loudly. For anything custom (structs, lists, +renames), implement `RecordMapper`: + +```java +public interface RecordMapper extends Serializable { + Record map(ITuple tuple, Schema schema); +} +``` + +The mapper takes `ITuple`, which both a bolt's `Tuple` and a `TridentTuple` satisfy, so a mapper +can be shared if you also write to Iceberg from elsewhere. + +### Partitioned tables + +Partitioned tables need no extra configuration: the writer derives each record's partition from +the table's current `PartitionSpec` and keeps one open data file per partition (Iceberg's fanout +writer), so a single batch can span any number of partitions. + +```java +Schema schema = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "region", Types.StringType.get()), + Types.NestedField.required(3, "event_time", Types.TimestampType.withZone())); + +PartitionSpec spec = PartitionSpec.builderFor(schema) + .identity("region") + .day("event_time") + .build(); + +IcebergOptions options = new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events") + .withAutoCreate(schema, spec) + .build(); +``` + +The `PartitionSpec` above is only used when the table is auto-created; for an existing table the +spec stored in the catalog wins. A batch spread over many partitions opens many files at once — +group the stream by the partition columns (`fieldsGrouping`) to keep file counts down. + +## Splitting the sink: writer and committer + +`IcebergBolt` writes and commits in the same task, so a sink at parallelism N committing every T +seconds produces `N/T` snapshots per second — each a metadata rewrite plus a compare-and-swap on +the catalog. Past a certain parallelism the only way to keep that load down is to commit less +often, which is to say to accept more latency. + +`IcebergWriterBolt` and `IcebergCommitterBolt` break that coupling. Writers seal batches on the +same thresholds `IcebergBolt` uses, but instead of committing they emit one descriptor tuple +carrying the batch's data files, **anchored to every tuple of the batch**, and then ack those +tuples. Anchoring is what preserves the guarantee: acking an input after emitting an anchored child +does not close the spout tuple's ack tree, so the source still advances only once the descriptor +itself is acked. A single committer accumulates descriptors from every writer, appends them all in +one Iceberg commit, and acks. Commit cost stops scaling with the sink's parallelism. + +```java +builder.setBolt("iceberg-writer", new IcebergWriterBolt(writerOptions), 8) + .fieldsGrouping("events", new Fields("region")); +builder.setBolt("iceberg-committer", new IcebergCommitterBolt(committerOptions), 1) + .globalGrouping("iceberg-writer"); +``` + +The committer must run at a parallelism of one behind a `globalGrouping`. It logs a warning if it +finds more than one task of its own component, because each task would commit independently — the +cost the split exists to remove. + +The trade-offs against the monolithic bolt: + +- **Wider failure blast radius.** A failed append fails every writer's batch in the group, not one + task's. The delivery guarantee is unchanged — at-least-once, atomic commits — but a single + catalog hiccup replays more work. +- **The committer is a single point of throughput.** It handles one tuple per writer seal, carrying + file metadata rather than data, so this is rarely the constraint; liveness is what to watch, via + `iceberg-oldest-pending-age-ms`. +- **Descriptors are not small.** They serialize each data file through Iceberg's + `ContentFileParser`, per-column metrics included, so a wide schema sealing many partitions at once + puts hundreds of KB on the wire per seal. `withGroupCommitMaxDataFiles` bounds what the committer + accumulates; nothing bounds a single writer's seal but its own commit thresholds. + +**The committer needs a tick tuple.** `withGroupCommitIntervalMillis` is evaluated only when a +descriptor arrives, so after a lull one accumulated descriptor sits uncommitted indefinitely, its +tuples un-acked with it, until either another descriptor or a tick arrives. Configure +`Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS` for the topology, or `withTickIntervalSecs` on the +committer's own options. The same applies to the writers' `withCommitIntervalMillis` and to +`IcebergBolt`'s. + +## How a commit is made recoverable + +The window between "the data files are durable" and "the table references them" is the only place +a crash can do damage, and a write-ahead log closes it. + +1. The batch's data files are closed and become durable. Nothing is visible to readers yet. +2. A WAL entry naming those files is written under + `<table location>/metadata/_storm_wal/<topologyName>/<componentId>/<taskIndex>/`, through the + table's own `FileIO`, carrying a freshly minted commit id. It lives with the table, not on + worker-local disk, so a task relaunched on another host still finds it. +3. The files are appended in a single Iceberg operation that stamps that commit id on the + resulting snapshot's summary (`storm.iceberg.commit-id`). +4. The WAL entry is deleted, and only then are the tuples acked. + +The WAL belongs to whichever component commits: `IcebergBolt`, or `IcebergCommitterBolt` in the +split sink. `IcebergWriterBolt` writes no WAL entries, because it makes nothing visible. + +On startup, before writing anything new, each committing task settles whatever its previous +incarnation left behind. For every pending entry it asks the table whether a snapshot carries that +commit id: if one does, the commit landed and the entry is simply dropped; if none does, the commit never landed +and the data files — still durable — are appended again. The table itself answers the question, so +no identity from the source is needed. + +### When a commit fails + +A failed commit is resolved immediately, while the batch is still in hand, rather than left to the +next startup. The sink asks the table whether the commit landed: + +- **It landed** — the append reported an error but the snapshot carries the commit id, the classic + `CommitStateUnknownException`. The batch is visible, so its tuples are acked and the WAL entry + is dropped. No replay, no duplicates. +- **It did not land** — the WAL entry is dropped *before* the tuples are failed. The source + replays them and they are written exactly once; the abandoned data files become orphans. Were + the entry left in place, the next startup would append those files as well, duplicating the rows + the replay had already written. +- **The table cannot be reached** — the outcome is genuinely unknown, so the entry is left for + startup to settle. This is the only path that can produce duplicate rows from a failed commit. Review Comment: Fixed, in `docs/storm-iceberg.md` and in `README.md`. The "written exactly once" claim is gone from the did-not-land branch and from the caveats at 319-322. The replacement text names the unknown-state path as the reason the second answer is a sample rather than a verdict, instead of presenting it as a separate case that only applies when the table is unreachable. ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergWriter.java: ########## @@ -0,0 +1,192 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.common; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericAppenderFactory; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.TaskWriter; +import org.apache.iceberg.io.UnpartitionedWriter; +import org.apache.iceberg.util.PropertyUtil; +import org.apache.storm.tuple.ITuple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Turns tuples into durable Iceberg data files. Knows nothing about how they are delivered, so it + * serves any Storm API; making the files visible is {@link IcebergCommitter}'s job. + * + * <p>Files are written eagerly as tuples arrive and closed by {@link #complete()}. Until then + * nothing is visible to readers: an abandoned writer leaves orphan files, never partial rows. + */ +public class IcebergWriter implements Closeable { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergWriter.class); + private static final String CATALOG_NAME = "storm-iceberg"; + + private final IcebergOptions options; + private final int taskId; + + private Catalog catalog; + private Table table; + private TaskWriter<Record> writer; + private CountingAppenderFactory countingAppenderFactory; + + public IcebergWriter(IcebergOptions options, int taskId) { + this.options = options; + this.taskId = taskId; + } + + /** Build the catalog and load — or, when configured, create — the target table. */ + public void open() { + this.catalog = CatalogUtil.buildIcebergCatalog( + CATALOG_NAME, options.getCatalogProperties(), new Configuration()); + TableIdentifier identifier = TableIdentifier.parse(options.getTableIdentifier()); + this.table = loadOrCreateTable(identifier); + LOG.info("Opened Iceberg writer for table {}, task {}", identifier, taskId); + } + + private Table loadOrCreateTable(TableIdentifier identifier) { + if (options.getAutoCreateSchema() != null && !catalog.tableExists(identifier)) { + PartitionSpec spec = options.getAutoCreateSpec() == null + ? PartitionSpec.unpartitioned() + : options.getAutoCreateSpec(); + try { + LOG.info("Auto-creating Iceberg table {}", identifier); + return catalog.createTable(identifier, options.getAutoCreateSchema(), spec); + } catch (AlreadyExistsException e) { + LOG.info("Table {} was concurrently created by another task", identifier); + } + } + return catalog.loadTable(identifier); + } + + public Table table() { + return table; + } + + /** Map a tuple to a record and write it to the open file set. */ + public void write(ITuple tuple) throws IOException { + if (writer == null) { + writer = createWriter(); + } + writer.write(options.getRecordMapper().map(tuple, table.schema())); + } + + /** + * Close the open files and hand back what was written, leaving a fresh buffer behind. The + * files are durable but not yet referenced by the table. + */ + public List<DataFile> complete() throws IOException { + if (writer == null) { + return List.of(); + } + try { + return Arrays.asList(writer.complete().dataFiles()); + } finally { + writer = null; + resetBuffer(); + } + } + + /** Discard what has been written. Files already closed remain as orphans. */ + public void abort() { + if (writer == null) { + return; + } + try { + writer.abort(); + } catch (IOException e) { Review Comment: Confirmed against iceberg-core 1.11.0. `BaseTaskWriter.abort()` deletes through `Tasks.foreach(...).throwFailureWhenFinished().noRetry().run(file -> io.deleteFile(file.location()))`; `Tasks.run` without an exception class propagates unchecked, and `HadoopFileIO.deleteFile` wraps in `RuntimeIOException`. Now `catch (IOException | RuntimeException)`, which also covers the `close()` case you point at, so the catalog shutdown is reached and the client threads do not leak on rebalance. The `flush()`/`seal()` catch blocks are covered by the same change: they already catch `Exception`, but the `writer.abort()` call *inside* them was the escape route. Regression test `abortSwallowsAFileIoFailureSoTheBatchCanStillBeFailed` drives it through a `catalog-impl` whose `FileIO` fails every delete. I reverted the catch to `IOException` alone to confirm the test fails without the fix, with exactly the `RuntimeIOException` above. ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergCommitter.java: ########## @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.common; + +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Makes durable data files visible in an Iceberg table, atomically and recoverably. + * + * <p>A commit is prepared in the {@link CommitWal} first, then appended in a single Iceberg + * operation that stamps the commit id on the resulting snapshot, then cleared from the WAL. + * Because the append is atomic, readers never observe part of a batch. Because the snapshot + * carries the commit id, {@link #recover()} can tell a commit that landed from one that did not, + * without needing any identity from the source: the table itself answers the question. + * + * <p>This yields atomic commits with at-least-once delivery. A crash before the WAL entry exists + * leaves orphan data files, which are invisible to readers and removed by Iceberg's standard + * orphan-file maintenance; a replayed batch is written and committed again, and its rows stay + * visible until something downstream removes them. + * + * <p>One commit may cover many batches — the aggregated committer hands it the files of every + * writer it collected — but it is still a single atomic append carrying a single commit id. + */ +public class IcebergCommitter { + + public static final String COMMIT_ID_PROPERTY = "storm.iceberg.commit-id"; + /** + * How far before a WAL entry's own timestamp the snapshot scan still looks. The snapshot is + * always written after the entry, so only clock skew between the worker that wrote the entry + * and whatever stamped the snapshot's timestamp can put it earlier. + */ + static final long CLOCK_SKEW_ALLOWANCE_MS = TimeUnit.MINUTES.toMillis(10); + + private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class); + + private final Table table; + private final CommitWal wal; + private final IcebergMetrics metrics; + + public IcebergCommitter(Table table, CommitWal wal, IcebergMetrics metrics) { + this.table = table; + this.wal = wal; + this.metrics = metrics; + } + + /** + * Log, append and clear one commit. Does nothing when there is nothing to append. + * + * <p>Returns normally only when the batch is visible in the table — including the case where + * the append reported a failure but had in fact landed. Throwing means the batch is not + * visible and its tuples must be replayed. + */ + public void commit(List<DataFile> dataFiles) { + if (dataFiles.isEmpty()) { + return; + } + CommitWal.WalEntry entry = wal.write(dataFiles); + long startNanos = System.nanoTime(); + try { + append(entry, dataFiles); + } catch (RuntimeException e) { + settleFailedCommit(entry, dataFiles, startNanos, e); + return; + } + metrics.committed(dataFiles, System.nanoTime() - startNanos); + wal.delete(entry); Review Comment: Fixed at both sites, through a `deleteQuietly()` that logs instead of propagating. The append has already landed by the time this runs, so a delete failure was making the caller fail tuples that are visible in the table. A leftover entry is now harmless: startup discards it rather than appending it (see the recovery change on the `recover()` thread). Covered by `aLandedCommitSurvivesAFailureToDeleteItsWalEntry`. ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergWriter.java: ########## @@ -0,0 +1,192 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.common; + +import java.io.Closeable; +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericAppenderFactory; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.TaskWriter; +import org.apache.iceberg.io.UnpartitionedWriter; +import org.apache.iceberg.util.PropertyUtil; +import org.apache.storm.tuple.ITuple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Turns tuples into durable Iceberg data files. Knows nothing about how they are delivered, so it + * serves any Storm API; making the files visible is {@link IcebergCommitter}'s job. + * + * <p>Files are written eagerly as tuples arrive and closed by {@link #complete()}. Until then + * nothing is visible to readers: an abandoned writer leaves orphan files, never partial rows. + */ +public class IcebergWriter implements Closeable { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergWriter.class); + private static final String CATALOG_NAME = "storm-iceberg"; + + private final IcebergOptions options; + private final int taskId; + + private Catalog catalog; + private Table table; + private TaskWriter<Record> writer; + private CountingAppenderFactory countingAppenderFactory; + + public IcebergWriter(IcebergOptions options, int taskId) { + this.options = options; + this.taskId = taskId; + } + + /** Build the catalog and load — or, when configured, create — the target table. */ + public void open() { + this.catalog = CatalogUtil.buildIcebergCatalog( + CATALOG_NAME, options.getCatalogProperties(), new Configuration()); + TableIdentifier identifier = TableIdentifier.parse(options.getTableIdentifier()); + this.table = loadOrCreateTable(identifier); + LOG.info("Opened Iceberg writer for table {}, task {}", identifier, taskId); + } + + private Table loadOrCreateTable(TableIdentifier identifier) { + if (options.getAutoCreateSchema() != null && !catalog.tableExists(identifier)) { + PartitionSpec spec = options.getAutoCreateSpec() == null + ? PartitionSpec.unpartitioned() + : options.getAutoCreateSpec(); + try { + LOG.info("Auto-creating Iceberg table {}", identifier); + return catalog.createTable(identifier, options.getAutoCreateSchema(), spec); + } catch (AlreadyExistsException e) { + LOG.info("Table {} was concurrently created by another task", identifier); + } + } + return catalog.loadTable(identifier); + } + + public Table table() { + return table; + } + + /** Map a tuple to a record and write it to the open file set. */ + public void write(ITuple tuple) throws IOException { + if (writer == null) { + writer = createWriter(); + } + writer.write(options.getRecordMapper().map(tuple, table.schema())); + } + + /** + * Close the open files and hand back what was written, leaving a fresh buffer behind. The + * files are durable but not yet referenced by the table. + */ + public List<DataFile> complete() throws IOException { + if (writer == null) { + return List.of(); + } + try { + return Arrays.asList(writer.complete().dataFiles()); + } finally { + writer = null; + resetBuffer(); + } + } Review Comment: Half of this is right and I have taken it: the nulling moved out of the `finally`, so it happens only after `complete()` returns. A `complete()` that throws now leaves a writer the caller's `abort()` can act on. Pushing back on the other variant. Having `abort()` work off a saved reference would delete completed files after a *successful* `complete()` too, and that is the dangerous case rather than the harmless one: when the visibility check itself cannot reach the table the outcome is unknown, and the commit may in fact have landed. Deleting the files then leaves a snapshot referencing files that no longer exist — a corrupt table, not orphans. Orphaning is the right outcome under uncertainty, so the no-op after a successful `complete()` is deliberate now, with a comment saying why. `abortAfterACompletedBatchLeavesItsFilesAlone` pins the intended behaviour. ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/bolt/IcebergBolt.java: ########## @@ -0,0 +1,184 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.bolt; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.apache.iceberg.DataFile; +import org.apache.storm.Config; +import org.apache.storm.iceberg.common.CommitWal; +import org.apache.storm.iceberg.common.IcebergCommitter; +import org.apache.storm.iceberg.common.IcebergMetrics; +import org.apache.storm.iceberg.common.IcebergOptions; +import org.apache.storm.iceberg.common.IcebergWriter; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.topology.OutputFieldsDeclarer; +import org.apache.storm.topology.base.BaseTickTupleAwareRichBolt; +import org.apache.storm.tuple.Tuple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Appends tuples to an Apache Iceberg table, committing them in atomic batches. + * + * <p>Tuples are written to Iceberg data files as they arrive but are <em>not</em> acked until the + * commit that makes them visible has landed. A batch therefore either becomes visible in full and + * is acked, or is failed and replayed by the source. Readers never see part of a batch. + * + * <p>The guarantee is <strong>atomic commits with at-least-once delivery</strong>. A replayed + * batch is written again, and because the table is append-only with no equality deletes, the + * duplicate rows stay visible until something downstream removes them. A crash between writing the + * files and committing them leaves orphan data files: invisible to readers, and cleaned up by + * Iceberg's standard orphan-file maintenance, which this module does not run for you. + * + * <p>Batches are closed when any configured threshold is crossed — records, bytes, or, if tick + * tuples are configured, elapsed time. Because nothing is acked early, a larger batch costs + * latency and replay volume, not durability. + */ +public class IcebergBolt extends BaseTickTupleAwareRichBolt { + + private static final long serialVersionUID = 1L; + private static final Logger LOG = LoggerFactory.getLogger(IcebergBolt.class); + + private final IcebergOptions options; + + private transient OutputCollector collector; + private transient IcebergWriter writer; + private transient IcebergCommitter committer; + private transient IcebergMetrics metrics; + private transient List<Tuple> pending; + private transient long batchStartNanos; + + public IcebergBolt(IcebergOptions options) { + this.options = options; + } + + @Override + public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) { + this.collector = collector; + this.pending = new ArrayList<>(); + this.metrics = new IcebergMetrics(context); + int taskId = context.getThisTaskId(); + this.writer = new IcebergWriter(options, taskId); + writer.open(); + String topologyName = String.valueOf(topoConf.get(Config.TOPOLOGY_NAME)); + CommitWal wal = new CommitWal(writer.table(), topologyName, + context.getThisComponentId(), context.getThisTaskIndex()); + this.committer = new IcebergCommitter(writer.table(), wal, metrics); + // Settle whatever an earlier run of this task left half-committed, before writing anything + // new. A commit that never became visible is replayed here; one that did is dropped. + int replayed = committer.recover(); + if (replayed > 0) { + LOG.info("Replayed {} commit(s) left pending by an earlier run of global task id {} ({}/{})", + replayed, taskId, context.getThisComponentId(), context.getThisTaskIndex()); + } + } + + @Override + protected void process(Tuple tuple) { + try { + if (pending.isEmpty()) { + // Schema and partition spec evolution is picked up between batches, not mid-batch: + // every file in one commit is written against the same metadata. + writer.refreshTable(); + batchStartNanos = System.nanoTime(); + } + writer.write(tuple); + } catch (Exception e) { + LOG.error("Failed writing tuple to Iceberg, failing the open batch", e); + failBatch(); + // The try block above never adds to pending, so failBatch() provably did not cover + // this tuple: fail it here too, or it would only be replayed on timeout. + collector.fail(tuple); + return; Review Comment: Confirmed, and deliberately not addressed in this round. A permanent mapping error is indistinguishable from a transient IO error at this point, and the batch cannot make forward progress. The fix is an API decision rather than a bug fix — an opt-in skip-and-count, a dead-letter stream, or documentation only — so I would rather agree the shape than add surface unilaterally. My inclination is an opt-in `withSkipUnmappableTuples` that acks the tuple and increments a counter, leaving the default loud. Do you have a preference, or would you rather this went to the dev list? ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergCommitter.java: ########## @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.common; + +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Makes durable data files visible in an Iceberg table, atomically and recoverably. + * + * <p>A commit is prepared in the {@link CommitWal} first, then appended in a single Iceberg + * operation that stamps the commit id on the resulting snapshot, then cleared from the WAL. + * Because the append is atomic, readers never observe part of a batch. Because the snapshot + * carries the commit id, {@link #recover()} can tell a commit that landed from one that did not, + * without needing any identity from the source: the table itself answers the question. + * + * <p>This yields atomic commits with at-least-once delivery. A crash before the WAL entry exists + * leaves orphan data files, which are invisible to readers and removed by Iceberg's standard + * orphan-file maintenance; a replayed batch is written and committed again, and its rows stay + * visible until something downstream removes them. + * + * <p>One commit may cover many batches — the aggregated committer hands it the files of every + * writer it collected — but it is still a single atomic append carrying a single commit id. + */ +public class IcebergCommitter { + + public static final String COMMIT_ID_PROPERTY = "storm.iceberg.commit-id"; + /** + * How far before a WAL entry's own timestamp the snapshot scan still looks. The snapshot is + * always written after the entry, so only clock skew between the worker that wrote the entry + * and whatever stamped the snapshot's timestamp can put it earlier. + */ + static final long CLOCK_SKEW_ALLOWANCE_MS = TimeUnit.MINUTES.toMillis(10); + + private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class); + + private final Table table; + private final CommitWal wal; + private final IcebergMetrics metrics; + + public IcebergCommitter(Table table, CommitWal wal, IcebergMetrics metrics) { + this.table = table; + this.wal = wal; + this.metrics = metrics; + } + + /** + * Log, append and clear one commit. Does nothing when there is nothing to append. + * + * <p>Returns normally only when the batch is visible in the table — including the case where + * the append reported a failure but had in fact landed. Throwing means the batch is not + * visible and its tuples must be replayed. + */ + public void commit(List<DataFile> dataFiles) { + if (dataFiles.isEmpty()) { + return; + } + CommitWal.WalEntry entry = wal.write(dataFiles); + long startNanos = System.nanoTime(); + try { + append(entry, dataFiles); + } catch (RuntimeException e) { + settleFailedCommit(entry, dataFiles, startNanos, e); + return; + } + metrics.committed(dataFiles, System.nanoTime() - startNanos); + wal.delete(entry); + } + + /** + * Resolve a failed commit while the batch is still in hand, rather than leaving it to the next + * startup. Asking the table whether the commit landed turns an unknown outcome into a known + * one at the only moment when it can still be acted on. + * + * <p>If it landed, the batch is visible and the caller may ack. If it did not, the entry is + * dropped before the exception propagates: the caller will fail those tuples, the source will + * replay them, and a WAL entry left behind would make the next startup append the original + * files too — duplicating what the replay writes. The abandoned files become orphans instead, + * which is what orphan-file maintenance is for. + */ + private void settleFailedCommit(CommitWal.WalEntry entry, List<DataFile> dataFiles, + long startNanos, RuntimeException failure) { + boolean landed; + try { + landed = isVisible(entry); + } catch (RuntimeException e) { + // The table cannot be reached, so the outcome stays unknown. Leave the entry: startup + // will settle it, and replaying a commit is recoverable in a way that losing it is not. + metrics.commitFailed(); + failure.addSuppressed(e); + throw failure; + } + wal.delete(entry); + if (landed) { + // The data is visible, so it counts as committed however the append reported itself. + metrics.committed(dataFiles, System.nanoTime() - startNanos); + LOG.warn("Commit {} reported a failure but its snapshot is present; " + + "treating it as successful", entry.commitId(), failure); + return; + } + metrics.commitFailed(); + LOG.error("Commit {} did not land; its data files are left as orphans and its tuples " + + "will be replayed", entry.commitId(), failure); + throw failure; + } + + /** + * Settle every commit this task prepared but did not finish, replaying the ones whose snapshot + * never appeared and dropping the ones that are already visible. + * + * @return how many commits had to be replayed + */ + public int recover() { + int replayed = 0; + for (CommitWal.WalEntry entry : wal.listPending()) { + if (isVisible(entry)) { + LOG.info("Commit {} is already visible; dropping its WAL entry", entry.commitId()); + } else { + List<DataFile> dataFiles = wal.read(entry); + LOG.info("Commit {} never became visible; replaying {} data files", + entry.commitId(), dataFiles.size()); + long startNanos = System.nanoTime(); + try { + append(entry, dataFiles); Review Comment: Accepted, and the behaviour has changed rather than only the design notes. `recover()` now abandons every pending entry and lets its data files orphan. The reasoning is yours: an entry only survives to startup if the task died before its commit resolved, and in that case the batch was never acked, so the source replays it — appending here can only add a second copy of the rows that replay writes. For a reliable source this path could never prevent loss, only add duplicates. Two of your other findings go away with it, by construction rather than by patch: startup no longer appends anything, so there is no unbounded replay from `prepare()` and no dependence on a snapshot that `expire_snapshots` may have removed. The javadoc on `recover()` now carries the argument. `recoveryAbandonsAPreparedCommitThatNeverBecameVisible` and the two bolt-level tests were rewritten to assert the new behaviour. ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergCommitter.java: ########## @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.storm.iceberg.common; + +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Makes durable data files visible in an Iceberg table, atomically and recoverably. + * + * <p>A commit is prepared in the {@link CommitWal} first, then appended in a single Iceberg + * operation that stamps the commit id on the resulting snapshot, then cleared from the WAL. + * Because the append is atomic, readers never observe part of a batch. Because the snapshot + * carries the commit id, {@link #recover()} can tell a commit that landed from one that did not, + * without needing any identity from the source: the table itself answers the question. + * + * <p>This yields atomic commits with at-least-once delivery. A crash before the WAL entry exists + * leaves orphan data files, which are invisible to readers and removed by Iceberg's standard + * orphan-file maintenance; a replayed batch is written and committed again, and its rows stay + * visible until something downstream removes them. + * + * <p>One commit may cover many batches — the aggregated committer hands it the files of every + * writer it collected — but it is still a single atomic append carrying a single commit id. + */ +public class IcebergCommitter { + + public static final String COMMIT_ID_PROPERTY = "storm.iceberg.commit-id"; + /** + * How far before a WAL entry's own timestamp the snapshot scan still looks. The snapshot is + * always written after the entry, so only clock skew between the worker that wrote the entry + * and whatever stamped the snapshot's timestamp can put it earlier. + */ + static final long CLOCK_SKEW_ALLOWANCE_MS = TimeUnit.MINUTES.toMillis(10); + + private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitter.class); + + private final Table table; + private final CommitWal wal; + private final IcebergMetrics metrics; + + public IcebergCommitter(Table table, CommitWal wal, IcebergMetrics metrics) { + this.table = table; + this.wal = wal; + this.metrics = metrics; + } + + /** + * Log, append and clear one commit. Does nothing when there is nothing to append. + * + * <p>Returns normally only when the batch is visible in the table — including the case where + * the append reported a failure but had in fact landed. Throwing means the batch is not + * visible and its tuples must be replayed. + */ + public void commit(List<DataFile> dataFiles) { + if (dataFiles.isEmpty()) { + return; + } + CommitWal.WalEntry entry = wal.write(dataFiles); + long startNanos = System.nanoTime(); + try { + append(entry, dataFiles); + } catch (RuntimeException e) { + settleFailedCommit(entry, dataFiles, startNanos, e); + return; + } + metrics.committed(dataFiles, System.nanoTime() - startNanos); + wal.delete(entry); + } + + /** + * Resolve a failed commit while the batch is still in hand, rather than leaving it to the next + * startup. Asking the table whether the commit landed turns an unknown outcome into a known + * one at the only moment when it can still be acted on. + * + * <p>If it landed, the batch is visible and the caller may ack. If it did not, the entry is + * dropped before the exception propagates: the caller will fail those tuples, the source will + * replay them, and a WAL entry left behind would make the next startup append the original + * files too — duplicating what the replay writes. The abandoned files become orphans instead, + * which is what orphan-file maintenance is for. + */ + private void settleFailedCommit(CommitWal.WalEntry entry, List<DataFile> dataFiles, + long startNanos, RuntimeException failure) { + boolean landed; + try { + landed = isVisible(entry); + } catch (RuntimeException e) { + // The table cannot be reached, so the outcome stays unknown. Leave the entry: startup + // will settle it, and replaying a commit is recoverable in a way that losing it is not. + metrics.commitFailed(); + failure.addSuppressed(e); + throw failure; + } + wal.delete(entry); + if (landed) { + // The data is visible, so it counts as committed however the append reported itself. + metrics.committed(dataFiles, System.nanoTime() - startNanos); + LOG.warn("Commit {} reported a failure but its snapshot is present; " + + "treating it as successful", entry.commitId(), failure); + return; + } + metrics.commitFailed(); + LOG.error("Commit {} did not land; its data files are left as orphans and its tuples " + + "will be replayed", entry.commitId(), failure); + throw failure; + } + + /** + * Settle every commit this task prepared but did not finish, replaying the ones whose snapshot + * never appeared and dropping the ones that are already visible. + * + * @return how many commits had to be replayed + */ + public int recover() { + int replayed = 0; + for (CommitWal.WalEntry entry : wal.listPending()) { + if (isVisible(entry)) { + LOG.info("Commit {} is already visible; dropping its WAL entry", entry.commitId()); + } else { + List<DataFile> dataFiles = wal.read(entry); + LOG.info("Commit {} never became visible; replaying {} data files", + entry.commitId(), dataFiles.size()); + long startNanos = System.nanoTime(); + try { + append(entry, dataFiles); + } catch (RuntimeException e) { + metrics.commitFailed(); + throw e; + } Review Comment: Resolved by the recovery change discussed on the `append(entry, dataFiles)` thread: `prepare()` no longer commits anything, so an unreadable entry, an unresolvable spec, or a `FileIO` without prefix listing can no longer throw on every restart. There is nothing left to bound or quarantine. The `newAppend()` observation goes with it — nothing is re-appended, so files that `remove_orphan_files` has already deleted cannot produce dangling manifest entries. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
