Add storm-iceberg module: an Apache Iceberg sink bolt - #8950
Conversation
New external module writing Trident batches to Apache Iceberg tables directly from a topology, with exactly-once semantics. Each batch is committed in a single Iceberg transaction that atomically appends the data files and records the transaction id in the table property storm.trident.<topologyName>.<partitionIndex>.last-committed-txid, so replayed batches are detected and skipped even across worker crashes and commits whose outcome was unknown to the writer. Includes: - IcebergOptions: catalog properties passed verbatim to Iceberg's CatalogUtil.buildIcebergCatalog, so any catalog works with its standard keys; file format, target file size and table auto-creation. - RecordMapper with a field-name based default doing the standard primitive conversions; required columns without a value fail loudly. - Partitioned tables through Iceberg's fanout writer, one open data file per partition. - Per-batch table refresh, so schema and partition spec evolution is picked up without restarting the workers. - Metrics-v2 instrumentation: records written, data files and bytes committed, commit latency, commit failures, skipped replays. - A shutdown hook releasing the catalog client, since Trident's State has no lifecycle callback. - Documentation and two runnable example topologies, unpartitioned and partitioned.
…dent # Conflicts: # storm-dist/binary/final-package/src/main/assembly/binary.xml
…ingested per commit.
GGraziadei
changed the title
The Iceberg version was pinned in the module's own properties, where a project-wide dependency bump would not see it. It now lives in the root pom alongside hadoop.version, with the artifacts managed in dependencyManagement; storm-iceberg declares them without versions. This does not widen the dependency's reach. dependencyManagement fixes versions without adding dependencies, so modules that do not declare Iceberg still resolve none of it: storm-client and storm-server both show zero Iceberg artifacts. The direction also prevents it, since storm-iceberg depends on storm-client (provided) and nothing in core depends on storm-iceberg. Neither binary distribution bundles it either: like every other external/* connector, storm-iceberg ships only its README in both the full and the lite assembly, so Iceberg reaches only topologies that ask for it. The catalog implementations (iceberg-aws, -gcp, -hive, -nessie, ...) are deliberately absent from the managed set: they are supplied by the topology rather than pulled in transitively.
|
@GGraziadei just checking in. This is still a WIP, correct? Cheers |
|
Hi Rui, yes, I'm still working on this. The current commit works in principle, but I'm exploring an approach to reduce both the commit count and the ACK latency. Once that's sorted out, I'll move the PR to Ready . |
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 preserves the guarantee: acking an input after emitting an anchored child does not close the spout tuple's ack tree, so the source advances only once the descriptor is acked by the committer. A single committer accumulates descriptors and appends them in one Iceberg commit, so snapshot rate is set by the group-commit cadence rather than by writer parallelism. Measured on a 6-way sink at a 10 s cadence: 115 snapshots with IcebergBolt against 20 with the split sink, for the same 360k rows, the same 115 data files and the same visibility latency (p50 5,454 ms against 5,506 ms). Supporting changes: - DataFileCodec is extracted from CommitWal, since the descriptor tuples and the WAL now share one data-file serialisation. - The commit WAL is keyed by component and task index rather than task id, so a writer and a committer in the same topology cannot collide and a recovered task finds its own entries. - IcebergOptions gains group-commit thresholds (interval and max data files) and a per-component tick interval, so writer and committer can be paced independently. - Seal metrics and pending-commit gauges, so a stalled committer is visible. - Pending state stays visible for the duration of a commit, and write errors fail fast rather than being deferred to the seal. - A split-sink example topology, with both shapes documented.
rzo1
left a comment
rzo1
left a comment
There was a problem hiding this comment.
Short on time over the holidays, so this is a first pass only — more to follow.
Ran mvn verify -pl external/storm-iceberg,examples/storm-iceberg-examples on JDK 25: green, 62 tests, checkstyle and PMD clean. Nothing below blocks a merge from my side. Thanks for the very thorough writeup, it made the review a lot easier.
Details are inline. One thing that does not attach to a diff line: docs/index.md does not link the new page, so it will not show up alongside the other integrations.
| 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; | ||
| } |
There was a problem hiding this 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.
| - **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. |
There was a problem hiding this 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.
| } | ||
| try { | ||
| writer.abort(); | ||
| } catch (IOException e) { |
There was a problem hiding this 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.
| return; | ||
| } | ||
| metrics.committed(dataFiles, System.nanoTime() - startNanos); | ||
| wal.delete(entry); |
There was a problem hiding this 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.
| public List<DataFile> complete() throws IOException { | ||
| if (writer == null) { | ||
| return List.of(); | ||
| } | ||
| try { | ||
| return Arrays.asList(writer.complete().dataFiles()); | ||
| } finally { | ||
| writer = null; | ||
| resetBuffer(); | ||
| } | ||
| } |
There was a problem hiding this 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.
| 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; | ||
| } |
There was a problem hiding this 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.
| protected void onTickTuple(Tuple tuple) { | ||
| if (!pending.isEmpty()) { | ||
| flush(); | ||
| } | ||
| } |
There was a problem hiding this 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().
| 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 |
There was a problem hiding this 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.
| 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); |
There was a problem hiding this 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.
| } | ||
| return List.of(); | ||
| } | ||
| entries.sort(Comparator.comparing(WalEntry::location)); |
There was a problem hiding this comment.
Sorting by location string only orders correctly while epoch millis stay 13 digits.
3 participants
What is the purpose of the change
Adds
storm-iceberg, a sink that writes tuples from a Storm topology directlyinto an Apache Iceberg table, with no Kafka Connect or Spark job in between.
The guarantee is atomic commits with at-least-once delivery. A batch becomes
visible in one Iceberg append or not at all, so readers never see a partial
batch; tuples are acked only after the commit containing them has landed, so a
crash costs orphan data files and a replay rather than lost rows. Duplicates are
possible and are not removed — the sink is append-only and writes no equality
deletes — so they stay visible until something downstream dedupes.
Exactly-once is deliberately not claimed. It would require a deterministic
identity of the input, which comes from the source rather than the sink, and a
general-purpose module cannot assume every user has a replayable,
deterministically addressed source.
Two sink shapes
IcebergBoltwrites and commits in the same task. Simple, but a sink atparallelism N committing every T seconds produces N/T snapshots per second,
each a metadata rewrite plus a compare-and-swap on the catalog.
IcebergWriterBolt+IcebergCommitterBoltbreak that coupling. Writers sealbatches on the usual thresholds and emit one descriptor tuple carrying the
batch's data files, anchored to every tuple of the batch, then ack those
tuples; anchoring keeps the spout's ack tree open, so the source advances only
once the committer has committed. A single committer groups descriptors into
one Iceberg commit, so snapshot rate follows the group-commit cadence instead
of writer parallelism.
Measured on a 6-way sink at a 10 s cadence, same 360k rows: 115 snapshots
with
IcebergBolt, 20 with the split sink, the same 115 data files and thesame visibility latency (p50 5,454 ms vs 5,506 ms).
Commit recovery
Data files are made durable first, then a write-ahead log entry naming them is
written under the table's own metadata location with a freshly minted commit id,
then the files are appended in one operation that stamps that id on the snapshot
summary, then the entry is deleted and the tuples are acked. On startup a task
asks the table whether each pending commit id is present in a snapshot,
replaying only those that are absent — the table answers the question, so no
identity from the source is needed. A failed commit is settled in flight rather
than left to the next startup: if it landed, the tuples are acked; if it did
not, the WAL entry is dropped before the tuples are failed, so the replay writes
those rows exactly once.
Scope and packaging
merge-on-read are out of scope.
remove_orphan_files,rewrite_data_files,expire_snapshots) is out of scope but documented as still necessary.external/*connector it ships only its README, in both the full and the lite assembly.
iceberg-aws,-gcp,-hive,-nessie, ...) aresupplied by the topology, not pulled in transitively.
dependencyManagement, so aproject-wide dependency bump sees it.
What this is, and is not
An earlier revision of this PR exposed a Trident
State. It is now a bolt,following feedback on the dev list.
Benchmarking against a Kafka + Spark Structured Streaming path (same Iceberg
release, same REST catalog) found the two equivalent at matched commit
cadence — latency within 4 %, file count within 1 %, snapshot count identical
once the split sink is used. Claims for this module resting on latency or
throughput would not be supportable, and none are made here.
The one capability the indirect path does not have is closing a commit on
accumulated bytes rather than elapsed time. Under a 12:1 burst profile that
produced files whose median equalled their maximum (1,644 KB, CV 0.33) against
42–792 KB for the time-triggered job (CV 0.78, 18.8× spread), at 3.1× fewer
files — costing latency (p50 41 s vs 5.3 s). Where predictable table layout
under variable load matters more than freshness, that is the reason to use this
sink. Otherwise the indirect path remains the right choice.
How was the change tested
62 tests, all green, plus checkstyle and PMD, on JDK 25
(
mvn verify -pl external/storm-iceberg,examples/storm-iceberg-examples):IcebergCommitterTestIcebergOptionsTestIcebergCommitterBoltTestIcebergBoltTestFieldNameRecordMapperTestIcebergWriterBoltTestIcebergWriterTestCommitWalTestDataFileCodecTestIcebergMetricsTestIcebergSplitSinkTestThe tests run against a real Iceberg table (
HadoopCatalogover a JUnit@TempDir), not mocks of Iceberg. What they cover:IcebergGenericsand asserted — for both sink shapes, including the writerand committer wired together.
appeared is replayed on startup; one that is already visible is dropped
rather than appended twice.
successful and its tuples acked; one that did not land clears its WAL entry
before failing, so the source's replay writes those rows exactly once. Both
are exercised with a real table and only the append made flaky.
writer's input tuples, so a failed group commit fails the whole batch back to
the source.
a threshold commits and acks the whole batch, a tick tuple flushes a partial
batch, and a failed commit fails the buffered tuples instead of acking them.
buffered-byte accounting.
iceberg-commit-failuresincrement corresponds to replayed tuples.Beyond the unit tests, both sink shapes were run on a 2-supervisor Storm cluster
against Kafka and an Iceberg REST catalog, at up to 1.17M records per run, with
row counts, duplicate counts and file/snapshot profiles verified from the
Iceberg metadata after each run.
Three operational constraints found that way are documented in
docs/storm-iceberg.md, since each fails silently:topology.message.timeout.secs, or everytuple is replayed before its commit lands and then committed anyway.
are all retained until it is sealed.
file, so a task owning k partitions writes k files of about B/k.
Not covered, and worth knowing before merge:
catalogs, and S3 / object-store
FileIO, are untested here.optimistic retries, which were observed retrying successfully under load but
not deliberately stressed.
warehouse; on object storage the commit path is dominated by request latency
and the snapshot-count results in particular may differ.