Add storm-iceberg module: a Trident sink for Apache Iceberg

docs/storm-iceberg.md

@@ -0,0 +1,186 @@
+---
+title: Storm Apache Iceberg Integration
+layout: documentation
+documentation: true
+---
+
+Trident state implementation 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 exactly-once semantics.
+
+## 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")
+    .build();
+
+TridentTopology topology = new TridentTopology();
+topology.newStream("spout", spout)
+    .partitionPersist(new IcebergStateFactory(options),
+        new Fields("id", "name", "ts"),
+        new IcebergStateUpdater());
+```
+
+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.
+
+### 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 |
+| `withCommitIntervalBytes(long)` | disabled | Buffer batches until this many bytes, then commit once |
+| `withCommitIntervalMillis(long)` | disabled | Flush the buffered window once it is this old |
+
+### 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(TridentTuple tuple, Schema schema);
+}
+```
+
+### Partitioned tables
+
+Partitioned tables need no extra configuration: the state 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. Note that a batch spread over many partitions opens many files
+at once — partition the stream on the partition columns (`partitionBy`) to keep file counts down.
+
+### Commit batching (trades exactly-once for fewer snapshots)
+
+By default **every Trident batch is one Iceberg commit and one snapshot**. On a high-rate stream
+with small batches that produces thousands of snapshots and as many small files, which degrades
+metastore and reader planning.
+
+`withCommitIntervalBytes` and `withCommitIntervalMillis` buffer batches and commit them together:
+
+```java
+IcebergOptions options = new IcebergOptions.Builder()
+    .withCatalogProperties(catalogProps)
+    .withTable("db.events")
+    .withCommitIntervalBytes(128L * 1024 * 1024)
+    .withCommitIntervalMillis(60_000)
+    .build();
+```
+
+The writer stays open across the buffered batches, so data files also grow towards
+`write.target-file-size-bytes` instead of being one-file-per-batch. Whichever threshold is
+reached first flushes the window; setting neither keeps the default one-commit-per-batch.
+
+> **This weakens the delivery guarantee, and cannot be made not to.** Trident treats a batch as
+> delivered the moment `commit()` returns. Batches that are buffered have already been
+> acknowledged, so if the worker dies before the flush **they are lost** — not duplicated, lost.
+> Use it only where losing the last window on a crash is acceptable, and leave both settings
+> unset when you need the exactly-once guarantee described below.
+
+Two further consequences of buffering:
+
+- A batch that fails after earlier batches were buffered forces the whole window to be dropped,
+  since an open writer cannot un-write the failed attempt's records. Watch
+  `iceberg-windows-dropped`.
+- The time threshold is evaluated when the *next* batch is committed, so a stream that stops
+  entirely leaves its last window uncommitted until data resumes. This matches how
+  `TimedRotationPolicy` behaves for `HdfsState`; Trident delivers no tick tuples to states and
+  skips empty batches, so there is no in-state timer that could do better without introducing
+  concurrency on the commit path.
+
+### Metrics
+
+The state registers these metrics-v2 metrics per `partitionPersist` task, so they show up in
+whatever reporter the cluster is configured with:
+
+| Metric | Type | Meaning |
+|---|---|---|
+| `iceberg-records-written` | counter | Records handed to the Iceberg writer |
+| `iceberg-data-files-committed` | counter | Data files made visible by commits |
+| `iceberg-bytes-committed` | counter | Bytes in those data files |
+| `iceberg-commit-latency` | timer | Duration of the Iceberg commit transaction |
+| `iceberg-commit-failures` | counter | Commits that failed (including unknown state) |
+| `iceberg-batches-skipped` | counter | Replayed batches skipped as already committed |
+| `iceberg-batches-buffered` | counter | Batches held back waiting for the commit threshold |
+| `iceberg-windows-dropped` | counter | Buffered windows discarded; their batches were lost |
+
+A steadily rising `iceberg-data-files-committed` with a flat `iceberg-bytes-committed` is the
+small-files signature: consider fewer, larger batches or a compaction job.
+
+### Table refresh
+
+Each batch refreshes the table metadata before writing, so schema and partition spec evolution
+performed outside the topology is picked up without restarting the workers. This costs one
+catalog round-trip per batch on top of the commit's own.
+
+### Resource cleanup
+
+Trident's `State` has no shutdown callback, so the state registers a JVM shutdown hook to close
+the catalog (REST/HTTP client, Hive metastore connection, object store client) when the worker
+exits.
+
+## Examples
+
+`examples/storm-iceberg-examples` contains two runnable topologies writing to a local Hadoop
+catalog: `IcebergTridentExampleTopology` (unpartitioned) and
+`IcebergPartitionedTridentExampleTopology` (partitioned by `identity(region)` and
+`days(event_time)`).
+
+## Exactly-once semantics
+
+Each Trident batch is committed to Iceberg with a single atomic `Transaction` that both appends
+the batch's data files and records the transaction id in the table property
+`storm.trident.<topologyName>.<partitionIndex>.last-committed-txid`. When Trident replays a
+batch whose txid is already recorded, the state skips it. Because the txid marker and the data
+commit are atomic, this stays correct across worker crashes, replays, and commits whose outcome
+was unknown to the writer.
+
+### Caveats
+
+- Exactly-once requires a **transactional spout** (the same txid must carry the same batch
+  content on replay), exactly as for `HdfsState`.
+- **Resubmitting a topology from scratch** (fresh transactional state in ZooKeeper) restarts
+  Trident txids from low values, so markers from the previous run would wrongly skip batches.
+  Resubmit under a new topology name, or clear the `storm.trident.<topologyName>.*` table
+  properties first.
+- Each state partition performs its own Iceberg commit per batch. Iceberg resolves concurrent
+  append commits with optimistic retries (tune with the table property
+  `commit.retry.num-retries`), but beyond roughly 10–20 partitions consider reducing the
+  parallelism of the `partitionPersist`.
+- Batches that fail before commit can leave never-committed (invisible) data files behind.
+  Standard Iceberg maintenance (`remove_orphan_files`) cleans them up.

examples/pom.xml

@@ -36,6 +36,7 @@
         <module>storm-kafka-client-examples</module>
         <module>storm-jdbc-examples</module>
         <module>storm-hdfs-examples</module>
+        <module>storm-iceberg-examples</module>
         <module>storm-jms-examples</module>
         <module>storm-perf</module>
     </modules>

examples/storm-iceberg-examples/pom.xml

@@ -0,0 +1,99 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>storm-examples</artifactId>
+        <groupId>org.apache.storm</groupId>
+        <version>3.0.1-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>storm-iceberg-examples</artifactId>
+
+
+
+    <name>Storm Iceberg Examples</name>
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.storm</groupId>
+            <artifactId>storm-client</artifactId>
+            <version>${project.version}</version>
+            <scope>${provided.scope}</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.storm</groupId>
+            <artifactId>storm-iceberg</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-shade-plugin</artifactId>
+                <configuration>
+                    <createDependencyReducedPom>true</createDependencyReducedPom>
+                    <filters>
+                        <filter>
+                            <artifact>*:*</artifact>
+                            <excludes>
+                                <exclude>META-INF/*.SF</exclude>
+                                <exclude>META-INF/*.sf</exclude>
+                                <exclude>META-INF/*.DSA</exclude>
+                                <exclude>META-INF/*.dsa</exclude>
+                                <exclude>META-INF/*.RSA</exclude>
+                                <exclude>META-INF/*.rsa</exclude>
+                                <exclude>META-INF/*.EC</exclude>
+                                <exclude>META-INF/*.ec</exclude>
+                                <exclude>META-INF/MSFTSIG.SF</exclude>
+                                <exclude>META-INF/MSFTSIG.RSA</exclude>
+                            </excludes>
+                        </filter>
+                    </filters>
+                </configuration>
+                <executions>
+                    <execution>
+                        <phase>package</phase>
+                        <goals>
+                            <goal>shade</goal>
+                        </goals>
+                        <configuration>
+                            <transformers>
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
+                                </transformer>
+                            </transformers>
+                        </configuration>
+                    </execution>
+                </executions>
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-checkstyle-plugin</artifactId>
+                <!--Note - the version would be inherited-->
+            </plugin>
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-pmd-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>

examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/BoundedTupleSpout.java

@@ -0,0 +1,97 @@
+/*
+ * 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.examples;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.trident.operation.TridentCollector;
+import org.apache.storm.trident.spout.IBatchSpout;
+import org.apache.storm.trident.topology.MasterBatchCoordinator;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+/**
+ * Emits a bounded, deterministic sequence of generated tuples and then stops.
+ *
+ * <p>Each tuple's content is derived purely from its index, and the index range of a batch is
+ * derived purely from the batch id, so a replayed batch always carries the same tuples. That is
+ * what makes this spout usable with the exactly-once semantics of {@code IcebergState}: a
+ * cycling {@code FixedBatchSpout} would not be, and a fixed list cannot reach interesting volumes.
+ */
+public class BoundedTupleSpout implements IBatchSpout {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    private final long totalTuples;
+    private final int batchSize;
+    private final Fields outputFields;
+    private final ValuesFactory valuesFactory;
+
+    public BoundedTupleSpout(long totalTuples, int batchSize, Fields outputFields,
+                             ValuesFactory valuesFactory) {
+        this.totalTuples = totalTuples;
+        this.batchSize = batchSize;
+        this.outputFields = outputFields;
+        this.valuesFactory = valuesFactory;
+    }
+
+    @Override
+    public void open(Map<String, Object> conf, TopologyContext context) {
+    }
+
+    @Override
+    public void emitBatch(long batchId, TridentCollector collector) {
+        // Trident transaction ids start at MasterBatchCoordinator.INIT_TXID (1).
+        long start = (batchId - MasterBatchCoordinator.INIT_TXID) * batchSize;
+        if (start < 0 || start >= totalTuples) {
+            return;
+        }
+        long end = Math.min(start + batchSize, totalTuples);
+        for (long index = start; index < end; index++) {
+            collector.emit(valuesFactory.create(index));
+        }
+    }
+
+    @Override
+    public void ack(long batchId) {
+    }
+
+    @Override
+    public void close() {
+    }
+
+    @Override
+    public Map<String, Object> getComponentConfiguration() {
+        return new HashMap<>();
+    }
+
+    @Override
+    public Fields getOutputFields() {
+        return outputFields;
+    }
+
+    /** Builds the tuple at a given position in the sequence. */
+    public interface ValuesFactory extends Serializable {
+        Values create(long index);
+    }
+}