Add storm-iceberg module: an Apache Iceberg sink bolt

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.Map;
+import org.apache.storm.spout.SpoutOutputCollector;
+import org.apache.storm.task.TopologyContext;
+import org.apache.storm.topology.OutputFieldsDeclarer;
+import org.apache.storm.topology.base.BaseRichSpout;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+/**
+ * Emits a bounded sequence of generated tuples and then stops, replaying any tuple that fails.
+ *
+ * <p>Each tuple's content is derived purely from its index, and the index is used as its message
+ * id, so a failed tuple is re-emitted with exactly the same content. That makes the source
+ * replayable, which is what {@code IcebergBolt}'s at-least-once delivery needs: a failed batch is
+ * written again rather than lost.
+ *
+ * <p>Replay is what at-least-once means in practice here — a replayed tuple is appended a second
+ * time and both copies stay visible in the table, since the sink writes no equality deletes.
+ */
+public class BoundedTupleSpout extends BaseRichSpout {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    private final long totalTuples;
+    private final Fields outputFields;
+    private final ValuesFactory valuesFactory;
+
+    private transient SpoutOutputCollector collector;
+    private transient long nextIndex;
+    private transient int taskCount;
+    private transient int taskIndex;
+
+    public BoundedTupleSpout(long totalTuples, Fields outputFields, ValuesFactory valuesFactory) {
+        this.totalTuples = totalTuples;
+        this.outputFields = outputFields;
+        this.valuesFactory = valuesFactory;
+    }
+
+    @Override
+    public void open(Map<String, Object> conf, TopologyContext context, SpoutOutputCollector collector) {
+        this.collector = collector;
+        // Each task walks its own stride of the sequence, so the spout can be parallelised without
+        // any two tasks emitting the same row.
+        this.taskCount = context.getComponentTasks(context.getThisComponentId()).size();
+        this.taskIndex = context.getThisTaskIndex();
+        this.nextIndex = taskIndex;
+    }
+
+    @Override
+    public void nextTuple() {
+        if (nextIndex >= totalTuples) {
+            return;
+        }
+        long index = nextIndex;
+        nextIndex += taskCount;
+        collector.emit(valuesFactory.create(index), index);
+    }
+
+    @Override
+    public void fail(Object msgId) {
+        long index = (Long) msgId;
+        collector.emit(valuesFactory.create(index), index);
+    }
+
+    @Override
+    public void declareOutputFields(OutputFieldsDeclarer declarer) {
+        declarer.declare(outputFields);
+    }
+
+    /** Builds the tuple at a given position in the sequence. */
+    public interface ValuesFactory extends Serializable {
+        Values create(long index);
+    }
+}

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

@@ -0,0 +1,118 @@
+/*
+ * 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.util.HashMap;
+import java.util.Map;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
+import org.apache.storm.Config;
+import org.apache.storm.StormSubmitter;
+import org.apache.storm.iceberg.bolt.IcebergBolt;
+import org.apache.storm.iceberg.common.IcebergOptions;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.Values;
+
+/**
+ * Ingests 10 million generated rows into an Iceberg table on a local Hadoop catalog, committing
+ * roughly every 1 MB written rather than once per tuple.
+ *
+ * <p>Batching commits this way keeps the snapshot and file counts sane — at one commit per tuple,
+ * 10M rows would mean 10M snapshots — and costs nothing in durability: the bolt does not ack a
+ * tuple until the commit containing it has landed, so a worker that dies mid-batch has those
+ * tuples replayed by the spout rather than losing them.
+ *
+ * <p>A tick tuple every {@value #TICK_SECS} seconds bounds how long the last, partial batch can
+ * sit unwritten when the stream goes quiet. Without it a batch below the size threshold would
+ * wait for traffic that may never come.
+ *
+ * <p>The sink runs at {@value #SINK_PARALLELISM}-way parallelism: each task buffers and commits
+ * independently, sees roughly a quarter of the stream, and therefore takes about four times as
+ * many rows to cross the 1 MB threshold on its own. Expect roughly {@value #SINK_PARALLELISM}
+ * snapshots per flush round, not one.
+ *
+ * <p>The row count and the commit threshold can be overridden on the command line, which is the
+ * quick way to try the topology out:
+ * {@code <warehouse> <topologyName> <totalTuples> <commitIntervalBytes>}.
+ *
+ * <p>Run locally with:
+ * {@code storm local storm-iceberg-examples-*.jar
+ * org.apache.storm.iceberg.examples.IcebergBoltExampleTopology}
+ * then inspect the warehouse directory (default {@code file:///tmp/storm-iceberg-warehouse}).
+ */
+public final class IcebergBoltExampleTopology {
+
+    private static final long TOTAL_TUPLES = 10_000_000L;
+    private static final long COMMIT_INTERVAL_BYTES = 1024L * 1024;
+    private static final int SINK_PARALLELISM = 4;
+    private static final int SPOUT_PARALLELISM = 2;
+    private static final int TICK_SECS = 30;
+    private static final String[] WORDS =
+        {"storm", "iceberg", "bolt", "lakehouse", "parquet", "snapshot"};
+
+    private IcebergBoltExampleTopology() {
+    }
+
+    public static void main(String[] args) throws Exception {
+        String warehouse = args.length > 0 ? args[0] : "file:///tmp/storm-iceberg-warehouse";
+        String topologyName = args.length > 1 ? args[1] : "iceberg-example";
+        long totalTuples = args.length > 2 ? Long.parseLong(args[2]) : TOTAL_TUPLES;
+        long commitIntervalBytes =
+            args.length > 3 ? Long.parseLong(args[3]) : COMMIT_INTERVAL_BYTES;
+
+        Schema schema = new Schema(
+            Types.NestedField.required(1, "id", Types.LongType.get()),
+            Types.NestedField.required(2, "word", Types.StringType.get()));
+
+        Map<String, String> catalogProps = new HashMap<>();
+        catalogProps.put(CatalogUtil.ICEBERG_CATALOG_TYPE, CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP);
+        catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);
+
+        IcebergOptions options = new IcebergOptions.Builder()
+            .withCatalogProperties(catalogProps)
+            .withTable("example.words")
+            .withAutoCreate(schema, PartitionSpec.unpartitioned())
+            .withCommitIntervalBytes(commitIntervalBytes)
+            .build();
+
+        // A distinct word per row on purpose: a handful of repeated values would be dictionary
+        // encoded down to about a byte per row, and no realistic size threshold would ever be
+        // reached. High cardinality keeps the example representative of real ingestion.
+        BoundedTupleSpout spout = new BoundedTupleSpout(totalTuples,
+            new Fields("id", "word"),
+            index -> new Values(index, WORDS[(int) (index % WORDS.length)] + "-" + index));
+
+        TopologyBuilder builder = new TopologyBuilder();
+        builder.setSpout("words", spout, SPOUT_PARALLELISM);
+        builder.setBolt("iceberg", new IcebergBolt(options), SINK_PARALLELISM)
+            .shuffleGrouping("words");
+
+        Config conf = new Config();
+        conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, TICK_SECS);
+        // Un-acked tuples in flight. The sink holds a batch's tuples un-acked until it commits,
+        // so this bounds how much is replayed when a worker dies, and must stay comfortably above
+        // the number of tuples one batch accumulates.
+        conf.setMaxSpoutPending(200_000);
+        StormSubmitter.submitTopology(topologyName, conf, builder.createTopology());
+    }
+}