rzo1 commented on code in PR #8950: URL: https://github.com/apache/storm/pull/8950#discussion_r3769069979
########## 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: A single `isVisible()` check is not enough to conclude the commit did not land. `ErrorHandlers$CommitErrorHandler` maps HTTP 500/502/503/504 to `CommitStateUnknownException`, and `SnapshotProducer.commit()` only retries on `CommitFailedException`, so a 504 in front of a REST catalog can mean a commit the backend still applies afterwards. One immediate `refresh()` can miss that; by then the entry is deleted and the replay lands under a new commit id, leaving both copies visible. Worth knowing: `RESTTableOperations.reconcileOnSimpleUpdate` already does one refresh-and-check for snapshot-add-only updates, so this is a second sample — still immediate, still not a wait. A bounded re-check like `HiveTableOperations.checkCommitStatus` would narrow the window but cannot close it. ########## 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: `BaseTaskWriter.abort()` deletes completed files via `io::deleteFile` under `throwFailureWhenFinished()`, and `HadoopFileIO` wraps failures in the unchecked `RuntimeIOException` — so this catch does not cover it. With 400 tuples in `pending` and the NameNode in safe mode, that escapes `failBatch()` before `pending.forEach(collector::fail)`: the tuples are neither acked nor failed, they only replay on timeout, and the executor dies. Same in the `flush()`/`seal()` catch blocks, and in `close()`, where it also skips the catalog shutdown and leaks client threads on every rebalance. Most concrete bug I found. ########## 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: Outside the try, so a delete failure after a landed append propagates, the caller fails the batch, and an already-visible commit gets replayed. Same pattern at line 115. ########## 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: Both bullets are wrong given the unknown-state path. A commit that reports failure but lands afterwards is replayed by the source, so "written exactly once" does not hold, and this is not the only path that produces duplicates. Same claim again at lines 319-322. ########## 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: Worth arguing this one explicitly in the design notes. An entry only survives un-landed if the process died before the commit, and in that case the tuples were never acked, so the source replays them anyway — this append then adds the same rows a second time. For a reliable source the replay path cannot prevent loss, only add duplicates; dropping the entry and letting the files orphan would give strictly fewer. It does pay off for an unreliable spout, but that is not the documented setup. ########## 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; + } + pending.add(tuple); + metrics.recordsWritten(1); + if (shouldFlush()) { + flush(); + } + } + + @Override + protected void onTickTuple(Tuple tuple) { + if (!pending.isEmpty()) { + flush(); + } + } Review Comment: This only fires if the topology sets `TOPOLOGY_TICK_TUPLE_FREQ_SECS`. `IcebergBolt` has no `getComponentConfiguration()` override, unlike `IcebergWriterBolt` and `IcebergCommitterBolt`, so `withTickIntervalSecs` is silently dropped for this bolt — while `README.md:152-153` and `docs/storm-iceberg.md:183-184` read as if it applies. The options tables are worded correctly. Either add the override or reject the setting in `IcebergOptions.build()`. ########## 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: Nulling `writer` in the `finally` makes the `writer.abort()` in `IcebergBolt.flush()` and `IcebergWriterBolt.seal()` a no-op. It also applies when `complete()` succeeds and the commit then throws — the commoner failure — where `BaseTaskWriter` still holds files `abort()` would delete. Orphans only, but those calls clearly meant to do something. Nulling on the success path only, or having `abort()` work off a saved reference, fixes it. ########## 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: A poison tuple is unskippable here: `FieldNameRecordMapper` throws on a required column with no value, and `RecordMapper` forbids returning null. One tombstone against a required column takes the whole batch down, the source replays it, and the next batch dies at the same record — no forward progress, and a permanent mapping error is indistinguishable from a transient IO error. Same in `IcebergWriterBolt.process()`. Some documented escape hatch would help. ########## 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: No bound and no quarantine, and this runs from `prepare()`, so one permanently bad entry throws on every restart — unresolvable spec, corrupt WAL JSON, or a `FileIO` without prefix listing. Only remedy is deleting JSON by hand from the table metadata dir. Related: `newAppend()` does no existence check, so replaying files that `remove_orphan_files` already deleted commits successfully and leaves dangling manifest entries. The docs recommend that procedure as routine maintenance. ########## 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; + } + metrics.committed(dataFiles, System.nanoTime() - startNanos); + replayed++; + } + wal.delete(entry); + } + return replayed; + } + + /** + * The Iceberg append itself. Deliberately records no metrics: only the caller knows whether a + * thrown exception means the commit is absent or merely unconfirmed. + */ + private void append(CommitWal.WalEntry entry, List<DataFile> dataFiles) { + AppendFiles append = table.newAppend().set(COMMIT_ID_PROPERTY, entry.commitId()); + for (DataFile dataFile : dataFiles) { + append.appendFile(dataFile); + } + append.commit(); + } + + /** + * Whether a snapshot carrying this entry's commit id exists. + * + * <p>Only snapshots from the entry's own era are examined: a commit cannot have landed before + * the entry that describes it was written. On a table with a long history that skips most of + * the snapshot list, and it costs nothing in accuracy — an older snapshot could not carry this + * commit id, since the id is minted when the entry is written. + */ + private boolean isVisible(CommitWal.WalEntry entry) { + table.refresh(); + for (Snapshot snapshot : table.snapshots()) { + if (withinScanWindow(snapshot.timestampMillis(), entry.createdAtMs()) + && entry.commitId().equals(snapshot.summary().get(COMMIT_ID_PROPERTY))) { + return true; + } + } + return false; + } Review Comment: Once `expire_snapshots` removes the snapshot carrying the id, a landed commit reads as un-landed and gets appended again. Narrow under Iceberg defaults, since orphan cleanup (3d) drops the entry before snapshot expiry (5d) — but streaming tables often expire far more aggressively, exactly because this module writes one snapshot per commit. The inverse also holds: orphan cleanup can remove live WAL entries. ########## 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. + +Note what the WAL does and does not protect. It protects the *reference* to durable data, which is +why atomic commits survive a crash. It does not give exactly-once: a batch that failed before its +WAL entry existed is replayed from the source and written afresh, duplicates included. + +A crash before step 2 leaves orphan data files. They are invisible to readers, and cleaned up by +Iceberg's standard `remove_orphan_files` maintenance. + +### Upgrading from an earlier layout Review Comment: This describes a layout that never shipped, so no deployment can be on it. Suggest dropping the section, and the matching "Upgrade note" at `README.md:222`. ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/bolt/IcebergCommitterBolt.java: ########## @@ -0,0 +1,208 @@ +/* + * 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.HashMap; +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.DataFileCodec; +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; + +/** + * Makes the files produced by every {@link IcebergWriterBolt} visible in a single Iceberg commit. + * + * <p>Run with a parallelism of one and a {@code globalGrouping}: the point of this component is + * that commit cost stops scaling with the sink's parallelism, so a topology can commit an order of + * magnitude more often at the same load on the catalog. That, in turn, is what keeps ack latency + * below the message timeout without acking anything early. + * + * <p>Descriptors are acked only after the append is visible, so the source still replays a failed + * batch. The failure blast radius is wider than the monolithic sink's — a failed append replays + * every writer's batch in the group, not one task's — which is the price of the aggregation. + * + * <p>Throughput is not a concern here: this bolt sees one tuple per writer seal, carrying file + * descriptors rather than data. Liveness is, which is what the pending gauges are for. + */ +public class IcebergCommitterBolt extends BaseTickTupleAwareRichBolt { + + private static final long serialVersionUID = 1L; + private static final Logger LOG = LoggerFactory.getLogger(IcebergCommitterBolt.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> sealed; + private transient List<DataFile> pendingFiles; + // Read by the metrics thread through the pending gauges, written by the executor thread. + private transient volatile long groupStartNanos; + + public IcebergCommitterBolt(IcebergOptions options) { + this.options = options; + } + + @Override + public void prepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector collector) { + int committerTasks = context.getComponentTasks(context.getThisComponentId()).size(); + if (committerTasks > 1) { + LOG.warn("{} is running with a parallelism of {}: every task commits independently, so the " + + "table receives {} snapshots per interval instead of one. Set its parallelism to 1 " + + "and feed it with a globalGrouping.", + context.getThisComponentId(), committerTasks, committerTasks); + } + this.collector = collector; + this.sealed = new ArrayList<>(); + this.pendingFiles = new ArrayList<>(); + this.metrics = new IcebergMetrics(context); + // The writer is held only for its table handle and catalog lifecycle; it never writes here. + this.writer = new IcebergWriter(options, context.getThisTaskId()); + 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); + metrics.registerPendingGauges(context, () -> pendingFiles.size(), this::oldestPendingAgeMs); Review Comment: `pendingFiles` is read from the metrics thread while the executor mutates it. Harmless for `size()` — plain `int` read, cannot throw or tear — just inconsistent with the `volatile` on `groupStartNanos` right above. ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/CommitWal.java: ########## @@ -0,0 +1,169 @@ +/* + * 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 com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.databind.JsonNode; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.OutputStreamWriter; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.UUID; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.SupportsPrefixOperations; +import org.apache.iceberg.util.JsonUtil; + +/** + * Write-ahead log of Iceberg commits that have been prepared but not yet made visible. + * + * <p>An entry is written after the batch's data files are durable and before the Iceberg commit + * that references them. It therefore protects only the reference, not the data: a crash before the + * entry exists leaves orphan data files, a crash after it exists is recoverable, because the entry + * names the files and carries the commit id that + * {@link IcebergCommitter} records in the resulting snapshot's summary. + * + * <p>Entries live under the table's metadata location, not on worker-local disk, so a task + * relaunched on another host still finds the commits it left behind. + */ +public final class CommitWal { + + static final String WAL_DIR = "_storm_wal"; + private static final String COMMIT_ID = "commit-id"; + private static final String CREATED_AT_MS = "created-at-ms"; + private static final String DATA_FILES = "data-files"; + + private final Table table; + private final FileIO io; + private final String prefix; + + /** + * Entries are keyed by component and task index rather than by global task id: task ids are + * assigned per submission and shift when the topology's structure changes, which would strand + * an entry under an id nobody reads again. + */ + public CommitWal(Table table, String topologyName, String componentId, int taskIndex) { + this.table = table; + this.io = table.io(); + String location = table.location(); + while (location.endsWith("/")) { + location = location.substring(0, location.length() - 1); + } + this.prefix = location + "/metadata/" + WAL_DIR + "/" + topologyName + + "/" + componentId + "/" + taskIndex; + } + + /** Record the files of one prepared commit, returning the entry that identifies it. */ + public WalEntry write(List<DataFile> dataFiles) { + String commitId = UUID.randomUUID().toString(); + long createdAtMs = System.currentTimeMillis(); + // The creation time goes in the file name as well as the body, so listing the WAL yields + // it without opening anything. + String location = prefix + "/" + createdAtMs + "-" + commitId + ".json"; + OutputFile outputFile = io.newOutputFile(location); + try (OutputStream out = outputFile.create(); + JsonGenerator json = JsonUtil.factory() + .createGenerator(new OutputStreamWriter(out, StandardCharsets.UTF_8))) { + json.writeStartObject(); + json.writeStringField(COMMIT_ID, commitId); + json.writeNumberField(CREATED_AT_MS, createdAtMs); + json.writeFieldName(DATA_FILES); + DataFileCodec.writeArray(json, dataFiles, table); + json.writeEndObject(); + } catch (IOException e) { + throw new UncheckedIOException("Failed writing Iceberg commit WAL entry " + location, e); + } + return new WalEntry(commitId, location, createdAtMs); + } + + /** Entries left behind by this topology and task, oldest first. */ + public List<WalEntry> listPending() { + if (!(io instanceof SupportsPrefixOperations)) { + throw new UnsupportedOperationException( + "Iceberg FileIO " + io.getClass().getName() + " cannot list the commit WAL; " + + "use a FileIO supporting prefix operations"); + } + List<WalEntry> entries = new ArrayList<>(); + try { + ((SupportsPrefixOperations) io).listPrefix(prefix + "/") + .forEach(fileInfo -> { + String location = fileInfo.location(); + if (location.endsWith(".json")) { + entries.add(new WalEntry(commitIdOf(location), location, createdAtMsOf(location))); + } + }); + } catch (UncheckedIOException e) { + // A task that has never committed has no WAL directory. On a hierarchical file system + // listing it raises FileNotFoundException; on object stores the prefix is simply empty. + if (!(e.getCause() instanceof FileNotFoundException)) { + throw e; + } + return List.of(); + } + entries.sort(Comparator.comparing(WalEntry::location)); Review Comment: Sorting by location string only orders correctly while epoch millis stay 13 digits. ########## external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/CommitWal.java: ########## @@ -0,0 +1,169 @@ +/* + * 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 com.fasterxml.jackson.core.JsonGenerator; +import com.fasterxml.jackson.databind.JsonNode; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.io.OutputStreamWriter; +import java.io.UncheckedIOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.List; +import java.util.UUID; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.Table; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.SupportsPrefixOperations; +import org.apache.iceberg.util.JsonUtil; + +/** + * Write-ahead log of Iceberg commits that have been prepared but not yet made visible. + * + * <p>An entry is written after the batch's data files are durable and before the Iceberg commit + * that references them. It therefore protects only the reference, not the data: a crash before the + * entry exists leaves orphan data files, a crash after it exists is recoverable, because the entry + * names the files and carries the commit id that + * {@link IcebergCommitter} records in the resulting snapshot's summary. + * + * <p>Entries live under the table's metadata location, not on worker-local disk, so a task + * relaunched on another host still finds the commits it left behind. + */ +public final class CommitWal { + + static final String WAL_DIR = "_storm_wal"; + private static final String COMMIT_ID = "commit-id"; + private static final String CREATED_AT_MS = "created-at-ms"; + private static final String DATA_FILES = "data-files"; + + private final Table table; + private final FileIO io; + private final String prefix; + + /** + * Entries are keyed by component and task index rather than by global task id: task ids are + * assigned per submission and shift when the topology's structure changes, which would strand + * an entry under an id nobody reads again. + */ + public CommitWal(Table table, String topologyName, String componentId, int taskIndex) { + this.table = table; + this.io = table.io(); + String location = table.location(); + while (location.endsWith("/")) { + location = location.substring(0, location.length() - 1); + } + this.prefix = location + "/metadata/" + WAL_DIR + "/" + topologyName + + "/" + componentId + "/" + taskIndex; + } Review Comment: No cluster qualifier here. Two clusters running a same-named topology against one table share this prefix, and one side's `recover()` will replay and delete the other's entries. Leaving out the submission id is clearly deliberate and right; it is the deployment identity that is missing. Consequence is duplicates rather than loss, so still inside the contract, but a user-supplied namespace or a doc warning would help. -- 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]
