Add storm-iceberg module: an Apache Iceberg sink bolt

[GGraziadei]

What is the purpose of the change

Adds storm-iceberg, a sink that writes tuples from a Storm topology directly
into 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

IcebergBolt writes and commits in the same task. Simple, but 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.

IcebergWriterBolt + IcebergCommitterBolt break that coupling. Writers seal
batches 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 the
same 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

• Append-only, format-version-2 tables. Row-level deletes, upserts and
  merge-on-read are out of scope.
• Table maintenance (remove_orphan_files, rewrite_data_files,
  expire_snapshots) is out of scope but documented as still necessary.
• Not bundled in either binary distribution: like every other external/*
  connector it ships only its README, in both the full and the lite assembly.
• Catalog implementations (iceberg-aws, -gcp, -hive, -nessie, ...) are
  supplied by the topology, not pulled in transitively.
• The Iceberg version is pinned in the root pom's dependencyManagement, so a
  project-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):

Test class	Tests
IcebergCommitterTest	9
IcebergOptionsTest	8
IcebergCommitterBoltTest	8
IcebergBoltTest	7
FieldNameRecordMapperTest	6
IcebergWriterBoltTest	6
IcebergWriterTest	5
CommitWalTest	4
DataFileCodecTest	3
IcebergMetricsTest	3
IcebergSplitSinkTest	3

The tests run against a real Iceberg table (HadoopCatalog over a JUnit
@TempDir), not mocks of Iceberg. What they cover:

• End to end: tuples written, committed, then read back through
  IcebergGenerics and asserted — for both sink shapes, including the writer
  and committer wired together.
• Both crash windows of the WAL: a prepared commit whose snapshot never
  appeared is replayed on startup; one that is already visible is dropped
  rather than appended twice.
• Failed commits: a commit that reported an error but landed is treated as
  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.
• Anchoring: the committer's ack of a descriptor is what releases the
  writer's input tuples, so a failed group commit fails the whole batch back to
  the source.
• Bolt semantics: tuples are not acked before their commit lands, reaching
  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.
• Writer: auto-create, partitioned fanout, empty-batch handling,
  buffered-byte accounting.
• Metrics: counters follow the outcome rather than the exception, so every
  iceberg-commit-failures increment 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:

• The batch must be shorter than topology.message.timeout.secs, or every
  tuple is replayed before its commit lands and then committed anyway.
• Worker heap must scale with the byte threshold, since an open batch's tuples
  are all retained until it is sealed.
• On a partitioned table the threshold sizes a task's commit, not an individual
  file, so a task owning k partitions writes k files of about B/k.

Not covered, and worth knowing before merge:

• Only the Hadoop and REST catalogs are exercised. Hive, Glue and Nessie
  catalogs, and S3 / object-store FileIO, are untested here.
• No multi-worker failure injection: concurrent appends rely on Iceberg's own
  optimistic retries, which were observed retrying successfully under load but
  not deliberately stressed.
• Benchmark figures come from single runs on one host with a local filesystem
  warehouse; on object storage the commit path is dominated by request latency
  and the snapshot-count results in particular may differ.

[reiabreu]

@GGraziadei just checking in. This is still a WIP, correct? Cheers

[GGraziadei]

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 .