GGraziadei commented on code in PR #8950: URL: https://github.com/apache/storm/pull/8950#discussion_r3789326031
########## 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: Added `withWalNamespace(String)`, validated at build time as a single non-blank path segment and interpolated ahead of the topology name in the WAL prefix. Left unset the layout is unchanged. You are right that leaving out the submission id is deliberate and that deployment identity was the missing part; a namespace the topology supplies is the smallest thing that expresses it, since the sink cannot discover the cluster it is running in. `entriesAreIsolatedByNamespace` covers it, and the caveats section of the docs now warns about two clusters writing to one table. ########## 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: No longer reachable from startup: recovery does not consult the snapshot list at all now (see the `recover()` thread), so an aggressive `expire_snapshots` cannot make a landed commit read as un-landed at the next restart, and orphan cleanup removing a live WAL entry no longer costs anything. It still applies to the in-flight check, where the window between writing the entry and asking the table is seconds rather than days — no plausible retention policy closes that one. ########## 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: Confirmed and fixed: `IcebergBolt` now overrides `getComponentConfiguration()` like the other two bolts, so `withTickIntervalSecs` reaches it and it inherits the topology-wide setting when none is configured. `theBoltDeclaresItsOwnTickInterval` covers both cases. I went with the override rather than rejecting the setting in `IcebergOptions.build()`, since the same options object is shared between the bolts in the split sink and rejecting it there would be worse. `README.md:152-153` and `docs/storm-iceberg.md:183-184` are reworded so they no longer read as if this bolt were the exception. ########## 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: Removed, together with the matching "Upgrade note" at `README.md:222`. `git log --all` confirms the module has only ever existed on this branch, so no deployment can be on the old layout and the section could only mislead. ########## 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: Agreed on the analysis, and left as it is — but `volatile` would not actually be the consistent choice here. `pendingFiles` is mutated in place (`addAll`, `clear`) and never reassigned, so a volatile reference would publish nothing the gauge reads; the `size()` field it dereferences is a plain int inside the list either way. `groupStartNanos` is volatile because it *is* reassigned, which is a real difference rather than an oversight. Adding volatile here would suggest a safety it does not provide, so the benign race stays. Happy to add a comment stating that if the asymmetry is what stands out. ########## 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: True until epoch millis reach 14 digits, in November 2286. It is now decorative rather than load-bearing: recovery deletes every entry instead of replaying them in order, so nothing depends on the ordering except the readability of the log line. Left as it is. -- 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]
