diff --git a/docs/storm-iceberg.md b/docs/storm-iceberg.md new file mode 100644 index 0000000000..8a6151528e --- /dev/null +++ b/docs/storm-iceberg.md @@ -0,0 +1,249 @@ +--- +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 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 | + +### 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. + +## 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 + `/metadata/_storm_wal///`, 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. + +On startup, before writing anything new, each 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. + +## Table maintenance is still your job + +This module writes; it does not maintain. A production table needs, on a schedule and from +outside the topology: + +- **`remove_orphan_files`** — reclaims files left by crashes and failed batches. +- **`rewrite_data_files`** (compaction) — streaming ingestion produces more, smaller files than + batch ingestion, whatever the commit interval. +- **`expire_snapshots`** — one snapshot per commit adds up quickly. + +Skipping these degrades read planning and grows storage cost over time. Iceberg ships these as +catalog procedures and as Spark actions; use whichever your platform already runs. + +## Metrics + +The bolt registers these metrics-v2 metrics per 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 | +| `iceberg-commit-failures` | counter | Commits that did **not** become visible | + +The counters follow the outcome, not the exception: an append that reported an error but whose +data is visible counts under `iceberg-data-files-committed`, not `iceberg-commit-failures`. Every +increment of `iceberg-commit-failures` therefore corresponds to tuples that were failed and +replayed. + +A steadily rising `iceberg-data-files-committed` with a flat `iceberg-bytes-committed` is the +small-files signature: raise the commit interval, or schedule compaction. + +## Table refresh + +The table metadata is refreshed between batches, so schema and partition spec evolution performed +outside the topology is picked up without restarting the workers. Every file within one commit is +written against the same metadata. + +## Dependencies + +`storm-iceberg` is **not bundled in the binary distributions**, in line with the direction of +[#8819](https://github.com/apache/storm/pull/8819): it is a topology-side library, and the +distribution's `external/` directory is on no classpath. + +It depends on the Iceberg core and format artifacts only. **Catalog implementations are not pulled +in transitively** — a topology writing through Glue, Nessie, or JDBC adds that catalog itself, and +one writing to S3 adds `iceberg-aws` and the AWS SDK. This keeps the dependency footprint of the +module small and puts the choice of object-store and catalog bindings where it belongs. + +The Iceberg version is pinned explicitly rather than tracking the newest release, so the module +stays buildable against Storm's Java baseline. + +## Examples + +`examples/storm-iceberg-examples` contains two runnable topologies writing to a local Hadoop +catalog: `IcebergBoltExampleTopology` (unpartitioned) and `IcebergPartitionedBoltExampleTopology` +(partitioned by `identity(region)` and `days(event_time)`). + +## Caveats + +- The commit WAL needs a `FileIO` that supports prefix listing. Iceberg's `HadoopFileIO`, + `S3FileIO` and `ResolvingFileIO` all do; an exotic custom `FileIO` may not. +- Each task commits independently. Iceberg resolves concurrent append commits with optimistic + retries (tune with the table property `commit.retry.num-retries`), but beyond roughly 10–20 + concurrent writers consider reducing the bolt's parallelism. +- If the table cannot be reached at all when a commit fails, the outcome stays unknown and the WAL + entry is left for the next startup to settle. That is the one case where a commit may be + replayed on top of tuples the source also replayed, producing duplicate rows. It is the + deliberate choice: replaying a commit is recoverable, losing one is not. diff --git a/examples/pom.xml b/examples/pom.xml index 93e6b83029..f663713e7b 100644 --- a/examples/pom.xml +++ b/examples/pom.xml @@ -36,6 +36,7 @@ storm-kafka-client-examples storm-jdbc-examples storm-hdfs-examples + storm-iceberg-examples storm-jms-examples storm-perf diff --git a/examples/storm-iceberg-examples/pom.xml b/examples/storm-iceberg-examples/pom.xml new file mode 100644 index 0000000000..1002ee6aab --- /dev/null +++ b/examples/storm-iceberg-examples/pom.xml @@ -0,0 +1,99 @@ + + + + 4.0.0 + + + storm-examples + org.apache.storm + 3.0.1-SNAPSHOT + ../pom.xml + + + storm-iceberg-examples + + + + Storm Iceberg Examples + + + org.apache.storm + storm-client + ${project.version} + ${provided.scope} + + + org.apache.storm + storm-iceberg + ${project.version} + + + + + + + org.apache.maven.plugins + maven-shade-plugin + + true + + + *:* + + META-INF/*.SF + META-INF/*.sf + META-INF/*.DSA + META-INF/*.dsa + META-INF/*.RSA + META-INF/*.rsa + META-INF/*.EC + META-INF/*.ec + META-INF/MSFTSIG.SF + META-INF/MSFTSIG.RSA + + + + + + + package + + shade + + + + + + + + + + + + + org.apache.maven.plugins + maven-checkstyle-plugin + + + + org.apache.maven.plugins + maven-pmd-plugin + + + + diff --git a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/BoundedTupleSpout.java b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/BoundedTupleSpout.java new file mode 100644 index 0000000000..95c4596f79 --- /dev/null +++ b/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. + * + *

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. + * + *

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 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); + } +} diff --git a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergBoltExampleTopology.java b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergBoltExampleTopology.java new file mode 100644 index 0000000000..ba483f1ac6 --- /dev/null +++ b/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. + * + *

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. + * + *

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. + * + *

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. + * + *

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 }. + * + *

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 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()); + } +} diff --git a/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedBoltExampleTopology.java b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedBoltExampleTopology.java new file mode 100644 index 0000000000..bcfe218131 --- /dev/null +++ b/examples/storm-iceberg-examples/src/main/java/org/apache/storm/iceberg/examples/IcebergPartitionedBoltExampleTopology.java @@ -0,0 +1,125 @@ +/* + * 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.time.Instant; +import java.time.temporal.ChronoUnit; +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 a partitioned Iceberg table on a local Hadoop catalog, + * committing roughly every 1 MB written rather than once per tuple. + * + *

The table is partitioned by {@code identity(region)} and {@code days(event_time)}, so a + * single batch spans several partitions and the sink opens one data file per partition through + * the Iceberg fanout writer. The threshold covers all four partitions together, so expect roughly + * four files per commit. + * + *

Buffering costs latency and replay volume, not durability: tuples are not acked until the + * commit that contains them lands, so a worker that dies mid-batch has them replayed. A tick tuple + * every {@value #TICK_SECS} seconds bounds how long a partial batch waits when the stream stalls. + * + *

The sink runs at {@value #SINK_PARALLELISM}-way parallelism: each task buffers and commits + * independently against its own share of the stream, so the 1 MB threshold is crossed roughly + * {@value #SINK_PARALLELISM} times more slowly per task than it would be at parallelism 1. + * + *

The row count and the commit threshold can be overridden on the command line: + * {@code }. + * + *

Run locally with: + * {@code storm local storm-iceberg-examples-*.jar + * org.apache.storm.iceberg.examples.IcebergPartitionedBoltExampleTopology} + * then inspect the warehouse directory (default {@code file:///tmp/storm-iceberg-warehouse}): the + * data files are laid out under {@code example/events/data/region=.../event_time_day=...}. + */ +public final class IcebergPartitionedBoltExampleTopology { + + 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[] REGIONS = {"eu-west", "us-east"}; + + private IcebergPartitionedBoltExampleTopology() { + } + + 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-partitioned-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, "region", Types.StringType.get()), + Types.NestedField.required(3, "event_time", Types.TimestampType.withZone())); + + PartitionSpec spec = PartitionSpec.builderFor(schema) + .identity("region") + .day("event_time") + .build(); + + Map 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.events") + .withAutoCreate(schema, spec) + .withCommitIntervalBytes(commitIntervalBytes) + .build(); + + // Two regions x two days -> four partitions, written by every sink task. + Instant today = Instant.now(); + Instant yesterday = today.minus(1, ChronoUnit.DAYS); + BoundedTupleSpout spout = new BoundedTupleSpout(totalTuples, + new Fields("id", "region", "event_time"), + index -> new Values(index, + REGIONS[(int) (index % REGIONS.length)], + index % 4 < 2 ? yesterday : today)); + + TopologyBuilder builder = new TopologyBuilder(); + builder.setSpout("events", spout, SPOUT_PARALLELISM); + builder.setBolt("iceberg", new IcebergBolt(options), SINK_PARALLELISM) + .shuffleGrouping("events"); + + Config conf = new Config(); + conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS, TICK_SECS); + // Un-acked tuples in flight; must stay comfortably above what one batch accumulates, + // since the sink holds a batch's tuples un-acked until it commits. + conf.setMaxSpoutPending(200_000); + StormSubmitter.submitTopology(topologyName, conf, builder.createTopology()); + } +} diff --git a/external/pom.xml b/external/pom.xml index 137f92634f..d5dab214d9 100644 --- a/external/pom.xml +++ b/external/pom.xml @@ -35,6 +35,7 @@ storm-hdfs storm-hdfs-blobstore storm-hdfs-oci + storm-iceberg storm-jdbc storm-jms storm-kafka-client diff --git a/external/storm-iceberg/README.md b/external/storm-iceberg/README.md new file mode 100644 index 0000000000..902cde73d4 --- /dev/null +++ b/external/storm-iceberg/README.md @@ -0,0 +1,149 @@ +# Storm Iceberg + +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**. + +Readers never see a partial batch. Tuples are acked only once the commit containing them has +landed, so nothing is lost; a replayed batch is written again, and because the sink is append-only +and writes no equality deletes, those duplicate rows stay visible until something downstream +removes them. The target table must be append-only and format version 2. + +Full documentation: [`docs/storm-iceberg.md`](../../docs/storm-iceberg.md). + +## Getting the jars + +Like every other `external/*` connector, `storm-iceberg` is **not bundled in either binary +distribution** — only this README ships. It is a topology-side library: the bolt runs in the +workers, and the distribution's `external/` directory is not on any classpath (see +`get_classpath()` in `bin/storm.py`). + +### Option 1 — depend on it from the topology (recommended) + +```xml + + org.apache.storm + storm-iceberg + ${storm.version} + +``` + +Shade it into the topology jar and nothing has to be installed on the cluster at all. + +### Option 2 — put it on the worker classpath + +If you would rather keep the Iceberg and Hadoop dependency trees out of the topology jar, the +distribution ships a helper that resolves `storm-iceberg` and its runtime dependencies from +Maven Central into `$STORM_HOME/extlib`: + +```bash +$STORM_HOME/bin/storm-iceberg-fetch +``` + +That is `extlib`, not `extlib-daemon`: the bolt runs inside the workers, and `extlib` is on +both the worker and the daemon classpath. Run it on every host running workers. + +It detects the Storm version from `$STORM_HOME/RELEASE`. Useful options: + +```bash +# explicit version / target directory +bin/storm-iceberg-fetch --version 3.0.0 --dest /opt/storm/extlib + +# pass extra arguments through to Maven (internal mirror / offline repo) +bin/storm-iceberg-fetch -- -s /etc/maven/settings.xml +bin/storm-iceberg-fetch -- -Dmaven.repo.local=/srv/offline-repo -o +``` + +Maven must be available on the host running the script (it does not have to be installed on +the cluster nodes — you can run it once and copy the resulting jars to every worker host). +Resubmit the topology afterwards so the new classpath takes effect. + +Neither route brings in catalog implementations or object-store bindings, by design: writing to +S3 additionally needs `iceberg-aws` and the AWS SDK, and catalogs such as Glue, Nessie or JDBC +need their own artifacts. None of them are dependencies of this module. + +## Usage + +```java +Map 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. + +### 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 | + +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. + +### 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); +} +``` + +## Recovery + +Data files are made durable first, then a write-ahead log entry naming them is written under +`

/metadata/_storm_wal///` with a freshly minted commit id, +then the files are appended in a single Iceberg operation that stamps that commit id on the +snapshot summary, then the entry is deleted and the tuples are acked. + +On startup a task settles whatever it left pending: it asks the table whether a snapshot carries +each entry's commit id, dropping the entry if the commit landed and re-appending the files if it +did not. The table answers the question, so no identity from the source is required — which is +exactly why the guarantee is at-least-once rather than exactly-once. + +A crash before the WAL entry exists leaves orphan data files: invisible to readers, and removed by +Iceberg's `remove_orphan_files`. + +A commit that *fails* is settled straight away rather than at the next startup: the sink asks the +table whether it landed. If it did — an append that reported an error but whose snapshot is +present — the tuples are acked and nothing is replayed. If it did not, the entry is dropped before +the tuples are failed, so the source's replay writes those rows exactly once and the abandoned +files become orphans. Only when the table itself is unreachable is the entry left for startup, +which is the one path that can duplicate rows. + +## Maintenance + +This module writes; it does not maintain. Schedule `remove_orphan_files`, `rewrite_data_files` +and `expire_snapshots` from outside the topology — streaming ingestion produces more, smaller +files and far more snapshots than batch ingestion does. diff --git a/external/storm-iceberg/pom.xml b/external/storm-iceberg/pom.xml new file mode 100644 index 0000000000..996ff4c734 --- /dev/null +++ b/external/storm-iceberg/pom.xml @@ -0,0 +1,100 @@ + + + + 4.0.0 + + + storm-external + org.apache.storm + 3.0.1-SNAPSHOT + ../pom.xml + + + storm-iceberg + + Storm Iceberg + + + + org.apache.storm + storm-client + ${project.version} + ${provided.scope} + + + org.apache.iceberg + iceberg-api + + + org.apache.iceberg + iceberg-core + + + org.apache.iceberg + iceberg-data + + + org.apache.iceberg + iceberg-parquet + + + + org.apache.iceberg + iceberg-orc + + + org.apache.hadoop + hadoop-client-api + ${hadoop.version} + + + org.apache.hadoop + hadoop-client-runtime + ${hadoop.version} + + + org.mockito + mockito-junit-jupiter + test + + + + + + org.apache.maven.plugins + maven-jar-plugin + + + + test-jar + + + + + + org.apache.maven.plugins + maven-checkstyle-plugin + + + + org.apache.maven.plugins + maven-pmd-plugin + + + + diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/bolt/IcebergBolt.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/bolt/IcebergBolt.java new file mode 100644 index 0000000000..32010bc646 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/bolt/IcebergBolt.java @@ -0,0 +1,179 @@ +/* + * 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. + * + *

Tuples are written to Iceberg data files as they arrive but are not 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. + * + *

The guarantee is atomic commits with at-least-once delivery. 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. + * + *

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 pending; + private transient long batchStartNanos; + + public IcebergBolt(IcebergOptions options) { + this.options = options; + } + + @Override + public void prepare(Map 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, taskId); + 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 task {}", replayed, taskId); + } + } + + @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); + pending.add(tuple); + metrics.recordsWritten(1); + } catch (Exception e) { + LOG.error("Failed writing tuple to Iceberg, failing the open batch", e); + failBatch(); + return; + } + if (shouldFlush()) { + flush(); + } + } + + @Override + protected void onTickTuple(Tuple tuple) { + if (!pending.isEmpty()) { + flush(); + } + } + + private boolean shouldFlush() { + Integer intervalRecords = options.getCommitIntervalRecords(); + if (intervalRecords != null && pending.size() >= intervalRecords) { + return true; + } + Long intervalBytes = options.getCommitIntervalBytes(); + if (intervalBytes != null && writer.bufferedBytes() >= intervalBytes) { + return true; + } + Long intervalMillis = options.getCommitIntervalMillis(); + return intervalMillis != null + && TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - batchStartNanos) >= intervalMillis; + } + + /** + * Close the batch's files, commit them, and only then ack. Any failure fails the whole batch: + * the source replays it, which is where at-least-once comes from. + */ + private void flush() { + List committing = new ArrayList<>(pending); + pending.clear(); + try { + List dataFiles = writer.complete(); + committer.commit(dataFiles); + } catch (Exception e) { + LOG.error("Failed committing {} tuple(s) to Iceberg, failing them for replay", + committing.size(), e); + writer.abort(); + committing.forEach(collector::fail); + return; + } + committing.forEach(collector::ack); + } + + private void failBatch() { + writer.abort(); + pending.forEach(collector::fail); + pending.clear(); + } + + @Override + public void cleanup() { + // Whatever is still buffered was never acked, so it will be replayed; drop it rather than + // race a commit against shutdown. + if (pending != null && !pending.isEmpty()) { + failBatch(); + } + if (writer != null) { + writer.close(); + } + } + + @Override + public void declareOutputFields(OutputFieldsDeclarer declarer) { + // Terminal sink: nothing is emitted downstream. + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/CommitWal.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/CommitWal.java new file mode 100644 index 0000000000..5ba5c5cbd9 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/CommitWal.java @@ -0,0 +1,171 @@ +/* + * 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.ContentFileParser; +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. + * + *

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. + * + *

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; + + public CommitWal(Table table, String topologyName, int taskId) { + 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 + "/" + taskId; + } + + /** Record the files of one prepared commit, returning the entry that identifies it. */ + public WalEntry write(List 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.writeArrayFieldStart(DATA_FILES); + for (DataFile dataFile : dataFiles) { + ContentFileParser.toJson(dataFile, table.specs().get(dataFile.specId()), json); + } + json.writeEndArray(); + 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 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 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)); + return entries; + } + + /** The data files named by an entry, resolved against the spec they were written with. */ + public List read(WalEntry entry) { + InputFile inputFile = io.newInputFile(entry.location()); + List dataFiles = new ArrayList<>(); + try (InputStream in = inputFile.newStream()) { + JsonNode root = JsonUtil.mapper().readTree(in); + for (JsonNode node : root.get(DATA_FILES)) { + dataFiles.add((DataFile) ContentFileParser.fromJson(node, table.specs())); + } + } catch (IOException e) { + throw new UncheckedIOException("Failed reading Iceberg commit WAL entry " + entry.location(), e); + } + return dataFiles; + } + + /** Drop an entry whose commit is known to be visible. */ + public void delete(WalEntry entry) { + io.deleteFile(entry.location()); + } + + private static String commitIdOf(String location) { + String name = fileName(location); + return name.substring(name.indexOf('-') + 1); + } + + private static long createdAtMsOf(String location) { + String name = fileName(location); + return Long.parseLong(name.substring(0, name.indexOf('-'))); + } + + private static String fileName(String location) { + String name = location.substring(location.lastIndexOf('/') + 1); + return name.substring(0, name.length() - ".json".length()); + } + + /** + * A prepared commit: its id, as recorded in the snapshot summary, where it is logged, and when + * it was logged — which bounds how far back recovery has to look for its snapshot. + */ + public record WalEntry(String commitId, String location, long createdAtMs) { + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/CountingAppenderFactory.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/CountingAppenderFactory.java new file mode 100644 index 0000000000..49bbda8e67 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/CountingAppenderFactory.java @@ -0,0 +1,109 @@ +/* + * 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.ArrayList; +import java.util.List; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.deletes.EqualityDeleteWriter; +import org.apache.iceberg.deletes.PositionDeleteWriter; +import org.apache.iceberg.encryption.EncryptedOutputFile; +import org.apache.iceberg.io.DataWriter; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.io.FileAppenderFactory; +import org.apache.iceberg.io.OutputFile; + +/** + * Wraps a {@link FileAppenderFactory} and remembers every writer it hands out, so the state can + * ask how many bytes the currently buffered window has produced. + * + *

The figure is an estimate: {@link FileAppender#length()} reflects what the + * underlying format has flushed, and columnar formats such as Parquet keep a sizeable in-memory + * buffer before writing a row group. It therefore under-reports until a file is closed, which for + * a commit threshold only means committing slightly later than the configured size. + * + *

Closed writers are kept in the list on purpose: a rolled-over file still counts towards the + * bytes accumulated since the last commit. {@link #reset()} drops them when the window is flushed. + */ +class CountingAppenderFactory implements FileAppenderFactory { + + private final FileAppenderFactory delegate; + private final List> appenders = new ArrayList<>(); + private final List> dataWriters = new ArrayList<>(); + + CountingAppenderFactory(FileAppenderFactory delegate) { + this.delegate = delegate; + } + + /** Bytes written by every writer created since the last {@link #reset()}. */ + long estimatedBytes() { + long total = 0L; + for (FileAppender appender : appenders) { + total += appender.length(); + } + for (DataWriter dataWriter : dataWriters) { + total += dataWriter.length(); + } + return total; + } + + /** Forget the writers of the window that was just committed or aborted. */ + void reset() { + appenders.clear(); + dataWriters.clear(); + } + + @Override + public FileAppender newAppender(OutputFile outputFile, FileFormat format) { + FileAppender appender = delegate.newAppender(outputFile, format); + appenders.add(appender); + return appender; + } + + @Override + public FileAppender newAppender(EncryptedOutputFile outputFile, FileFormat format) { + FileAppender appender = delegate.newAppender(outputFile, format); + appenders.add(appender); + return appender; + } + + @Override + public DataWriter newDataWriter(EncryptedOutputFile file, FileFormat format, + StructLike partition) { + DataWriter dataWriter = delegate.newDataWriter(file, format, partition); + dataWriters.add(dataWriter); + return dataWriter; + } + + @Override + public EqualityDeleteWriter newEqDeleteWriter(EncryptedOutputFile file, + FileFormat format, StructLike partition) { + // The sink is append-only; delete writers are never requested. + return delegate.newEqDeleteWriter(file, format, partition); + } + + @Override + public PositionDeleteWriter newPosDeleteWriter(EncryptedOutputFile file, + FileFormat format, + StructLike partition) { + return delegate.newPosDeleteWriter(file, format, partition); + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/FieldNameRecordMapper.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/FieldNameRecordMapper.java new file mode 100644 index 0000000000..3f85de3707 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/FieldNameRecordMapper.java @@ -0,0 +1,99 @@ +/* + * 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.nio.ByteBuffer; +import java.time.Instant; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import java.util.Date; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.apache.storm.tuple.ITuple; + +/** + * Default {@link RecordMapper}: matches tuple fields to Iceberg columns by name. + * + *

Standard primitive conversions are applied (numeric widening, {@link Instant}/epoch-millis to + * timestamp types, byte[] to ByteBuffer). Values of other types — including java.time values that + * already match the Iceberg representation, and struct/list/map values — are passed through as-is. + * A required column with no tuple value fails fast with {@link IllegalArgumentException}. + */ +public class FieldNameRecordMapper implements RecordMapper { + + private static final long serialVersionUID = 1L; + + @Override + public Record map(ITuple tuple, Schema schema) { + GenericRecord record = GenericRecord.create(schema); + for (Types.NestedField field : schema.columns()) { + Object value = tuple.contains(field.name()) ? tuple.getValueByField(field.name()) : null; + if (value == null) { + if (field.isRequired()) { + throw new IllegalArgumentException( + "Tuple has no value for required Iceberg column '" + field.name() + "'"); + } + continue; + } + record.setField(field.name(), convert(field.type(), value)); + } + return record; + } + + private Object convert(Type type, Object value) { + switch (type.typeId()) { + case INTEGER: + return value instanceof Number ? ((Number) value).intValue() : value; + case LONG: + return value instanceof Number ? ((Number) value).longValue() : value; + case FLOAT: + return value instanceof Number ? ((Number) value).floatValue() : value; + case DOUBLE: + return value instanceof Number ? ((Number) value).doubleValue() : value; + case STRING: + return value instanceof CharSequence ? value.toString() : value; + case TIMESTAMP: + return convertTimestamp((Types.TimestampType) type, value); + case BINARY: + return value instanceof byte[] ? ByteBuffer.wrap((byte[]) value) : value; + default: + return value; + } + } + + private Object convertTimestamp(Types.TimestampType type, Object value) { + Instant instant; + if (value instanceof Instant) { + instant = (Instant) value; + } else if (value instanceof Date) { + instant = ((Date) value).toInstant(); + } else if (value instanceof Long) { + instant = Instant.ofEpochMilli((Long) value); + } else { + return value; // already an OffsetDateTime / LocalDateTime + } + return type.shouldAdjustToUTC() + ? OffsetDateTime.ofInstant(instant, ZoneOffset.UTC) + : LocalDateTime.ofInstant(instant, ZoneOffset.UTC); + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergCommitter.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergCommitter.java new file mode 100644 index 0000000000..6cca0c6268 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergCommitter.java @@ -0,0 +1,190 @@ +/* + * 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. + * + *

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. + * + *

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. + */ +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. + * + *

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 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. + * + *

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 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 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 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. + * + *

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; + } + + static boolean withinScanWindow(long snapshotTimestampMs, long entryCreatedAtMs) { + return snapshotTimestampMs >= entryCreatedAtMs - CLOCK_SKEW_ALLOWANCE_MS; + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergMetrics.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergMetrics.java new file mode 100644 index 0000000000..7e273bc14b --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergMetrics.java @@ -0,0 +1,84 @@ +/* + * 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.codahale.metrics.Counter; +import com.codahale.metrics.Timer; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.apache.iceberg.DataFile; +import org.apache.storm.task.IMetricsContext; + +/** + * Storm metrics-v2 instrumentation for the Iceberg sink. + * + *

When no {@link IMetricsContext} is available the metrics are still recorded, on unregistered + * instances, so callers never need a null check. + */ +public class IcebergMetrics { + + static final String RECORDS_WRITTEN = "iceberg-records-written"; + static final String DATA_FILES_COMMITTED = "iceberg-data-files-committed"; + static final String BYTES_COMMITTED = "iceberg-bytes-committed"; + static final String COMMIT_LATENCY = "iceberg-commit-latency"; + static final String COMMIT_FAILURES = "iceberg-commit-failures"; + + private final Counter recordsWritten; + private final Counter dataFilesCommitted; + private final Counter bytesCommitted; + private final Timer commitLatency; + private final Counter commitFailures; + + public IcebergMetrics(IMetricsContext metrics) { + this.recordsWritten = counter(metrics, RECORDS_WRITTEN); + this.dataFilesCommitted = counter(metrics, DATA_FILES_COMMITTED); + this.bytesCommitted = counter(metrics, BYTES_COMMITTED); + this.commitFailures = counter(metrics, COMMIT_FAILURES); + this.commitLatency = timer(metrics, COMMIT_LATENCY); + } + + private static Counter counter(IMetricsContext metrics, String name) { + Counter registered = metrics == null ? null : metrics.registerCounter(name); + // Fall back to an unregistered instance: instrumentation must never be able to break the + // sink, whatever the metrics context does or does not hand back. + return registered == null ? new Counter() : registered; + } + + private static Timer timer(IMetricsContext metrics, String name) { + Timer registered = metrics == null ? null : metrics.registerTimer(name); + return registered == null ? new Timer() : registered; + } + + public void recordsWritten(long count) { + recordsWritten.inc(count); + } + + /** Counts the files and bytes made visible by a successful commit. */ + public void committed(List dataFiles, long durationNanos) { + dataFilesCommitted.inc(dataFiles.size()); + for (DataFile dataFile : dataFiles) { + bytesCommitted.inc(dataFile.fileSizeInBytes()); + } + commitLatency.update(durationNanos, TimeUnit.NANOSECONDS); + } + + public void commitFailed() { + commitFailures.inc(); + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergOptions.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergOptions.java new file mode 100644 index 0000000000..d39177a3fd --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergOptions.java @@ -0,0 +1,216 @@ +/* + * 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.io.Serial; +import java.io.Serializable; +import java.util.HashMap; +import java.util.Map; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; + +/** + * Serializable configuration for the Iceberg sink. + * + *

The catalog properties are passed verbatim to + * {@code CatalogUtil.buildIcebergCatalog(...)}, so any Iceberg catalog (hive, hadoop, rest, + * glue, nessie, ...) can be configured with the same property keys documented by Iceberg. + */ +public class IcebergOptions implements Serializable { + + /** Batch size used when a topology configures no commit threshold of its own. */ + public static final int DEFAULT_COMMIT_INTERVAL_RECORDS = 1000; + + @Serial + private static final long serialVersionUID = 1L; + + private final Map catalogProperties; + private final String tableIdentifier; + private final RecordMapper recordMapper; + private final FileFormat fileFormat; + private final Long targetFileSizeBytes; + private final Schema autoCreateSchema; + private final PartitionSpec autoCreateSpec; + private final Long commitIntervalBytes; + private final Long commitIntervalMillis; + private final Integer commitIntervalRecords; + + private IcebergOptions(Builder builder) { + this.catalogProperties = builder.catalogProperties; + this.tableIdentifier = builder.tableIdentifier; + this.recordMapper = builder.recordMapper; + this.fileFormat = builder.fileFormat; + this.targetFileSizeBytes = builder.targetFileSizeBytes; + this.autoCreateSchema = builder.autoCreateSchema; + this.autoCreateSpec = builder.autoCreateSpec; + this.commitIntervalBytes = builder.commitIntervalBytes; + this.commitIntervalMillis = builder.commitIntervalMillis; + this.commitIntervalRecords = builder.commitIntervalRecords; + } + + public Map getCatalogProperties() { + return catalogProperties; + } + + public String getTableIdentifier() { + return tableIdentifier; + } + + public RecordMapper getRecordMapper() { + return recordMapper; + } + + public FileFormat getFileFormat() { + return fileFormat; + } + + public Long getTargetFileSizeBytes() { + return targetFileSizeBytes; + } + + public Long getCommitIntervalBytes() { + return commitIntervalBytes; + } + + public Long getCommitIntervalMillis() { + return commitIntervalMillis; + } + + public Integer getCommitIntervalRecords() { + return commitIntervalRecords; + } + + public Schema getAutoCreateSchema() { + return autoCreateSchema; + } + + public PartitionSpec getAutoCreateSpec() { + return autoCreateSpec; + } + + public static class Builder { + private Map catalogProperties; + private String tableIdentifier; + private RecordMapper recordMapper = new FieldNameRecordMapper(); + private FileFormat fileFormat = FileFormat.PARQUET; + private Long targetFileSizeBytes; + private Schema autoCreateSchema; + private PartitionSpec autoCreateSpec; + private Long commitIntervalBytes; + private Long commitIntervalMillis; + private Integer commitIntervalRecords; + + public Builder withCatalogProperties(Map properties) { + this.catalogProperties = properties == null ? null : new HashMap<>(properties); + return this; + } + + public Builder withTable(String identifier) { + this.tableIdentifier = identifier; + return this; + } + + public Builder withRecordMapper(RecordMapper mapper) { + this.recordMapper = mapper; + return this; + } + + public Builder withFileFormat(FileFormat format) { + this.fileFormat = format; + return this; + } + + public Builder withTargetFileSizeBytes(long bytes) { + this.targetFileSizeBytes = bytes; + return this; + } + + /** + * Create the table on first use when it does not exist, with the given schema and + * partition spec. A null spec means unpartitioned. + */ + public Builder withAutoCreate(Schema schema, PartitionSpec spec) { + this.autoCreateSchema = schema; + this.autoCreateSpec = spec; + return this; + } + + /** + * Close the batch once roughly this many bytes have been written. Sizing batches this way + * keeps data files near the table's target size and the snapshot count sane. + * + *

Buffering costs latency and replay volume, not durability: buffered tuples are not + * acked, so a worker that dies before the commit has them replayed rather than lost. + */ + public Builder withCommitIntervalBytes(long bytes) { + this.commitIntervalBytes = bytes; + return this; + } + + /** + * Close the batch once it has been open this long, evaluated when the next tuple arrives. + * Configure {@link org.apache.storm.Config#TOPOLOGY_TICK_TUPLE_FREQ_SECS} as well to bound + * the latency of the last batch when the stream stalls. + */ + public Builder withCommitIntervalMillis(long millis) { + this.commitIntervalMillis = millis; + return this; + } + + /** Close the batch once it holds this many tuples. */ + public Builder withCommitIntervalRecords(int records) { + this.commitIntervalRecords = records; + return this; + } + + public IcebergOptions build() { + if (catalogProperties == null || catalogProperties.isEmpty()) { + throw new IllegalStateException("Catalog properties must be specified."); + } + if (tableIdentifier == null || tableIdentifier.isBlank()) { + throw new IllegalStateException("Table identifier must be specified."); + } + if (recordMapper == null) { + throw new IllegalStateException("RecordMapper must not be null."); + } + if (fileFormat == null) { + throw new IllegalStateException("FileFormat must not be null."); + } + if (targetFileSizeBytes != null && targetFileSizeBytes <= 0) { + throw new IllegalStateException("Target file size must be positive."); + } + if (commitIntervalBytes != null && commitIntervalBytes <= 0) { + throw new IllegalStateException("Commit interval bytes must be positive."); + } + if (commitIntervalMillis != null && commitIntervalMillis <= 0) { + throw new IllegalStateException("Commit interval millis must be positive."); + } + if (commitIntervalRecords != null && commitIntervalRecords <= 0) { + throw new IllegalStateException("Commit interval records must be positive."); + } + if (commitIntervalBytes == null && commitIntervalMillis == null && commitIntervalRecords == null) { + // Without a threshold a batch would stay open until a tick tuple arrived, which + // costs unbounded latency on a topology that configured none. + this.commitIntervalRecords = DEFAULT_COMMIT_INTERVAL_RECORDS; + } + return new IcebergOptions(this); + } + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergWriter.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergWriter.java new file mode 100644 index 0000000000..cf60cefbdc --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/IcebergWriter.java @@ -0,0 +1,192 @@ +/* + * 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.io.Closeable; +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.GenericAppenderFactory; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.exceptions.AlreadyExistsException; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.TaskWriter; +import org.apache.iceberg.io.UnpartitionedWriter; +import org.apache.iceberg.util.PropertyUtil; +import org.apache.storm.tuple.ITuple; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Turns tuples into durable Iceberg data files. Knows nothing about how they are delivered, so it + * serves any Storm API; making the files visible is {@link IcebergCommitter}'s job. + * + *

Files are written eagerly as tuples arrive and closed by {@link #complete()}. Until then + * nothing is visible to readers: an abandoned writer leaves orphan files, never partial rows. + */ +public class IcebergWriter implements Closeable { + + private static final Logger LOG = LoggerFactory.getLogger(IcebergWriter.class); + private static final String CATALOG_NAME = "storm-iceberg"; + + private final IcebergOptions options; + private final int taskId; + + private Catalog catalog; + private Table table; + private TaskWriter writer; + private CountingAppenderFactory countingAppenderFactory; + + public IcebergWriter(IcebergOptions options, int taskId) { + this.options = options; + this.taskId = taskId; + } + + /** Build the catalog and load — or, when configured, create — the target table. */ + public void open() { + this.catalog = CatalogUtil.buildIcebergCatalog( + CATALOG_NAME, options.getCatalogProperties(), new Configuration()); + TableIdentifier identifier = TableIdentifier.parse(options.getTableIdentifier()); + this.table = loadOrCreateTable(identifier); + LOG.info("Opened Iceberg writer for table {}, task {}", identifier, taskId); + } + + private Table loadOrCreateTable(TableIdentifier identifier) { + if (options.getAutoCreateSchema() != null && !catalog.tableExists(identifier)) { + PartitionSpec spec = options.getAutoCreateSpec() == null + ? PartitionSpec.unpartitioned() + : options.getAutoCreateSpec(); + try { + LOG.info("Auto-creating Iceberg table {}", identifier); + return catalog.createTable(identifier, options.getAutoCreateSchema(), spec); + } catch (AlreadyExistsException e) { + LOG.info("Table {} was concurrently created by another task", identifier); + } + } + return catalog.loadTable(identifier); + } + + public Table table() { + return table; + } + + /** Map a tuple to a record and write it to the open file set. */ + public void write(ITuple tuple) throws IOException { + if (writer == null) { + writer = createWriter(); + } + writer.write(options.getRecordMapper().map(tuple, table.schema())); + } + + /** + * Close the open files and hand back what was written, leaving a fresh buffer behind. The + * files are durable but not yet referenced by the table. + */ + public List complete() throws IOException { + if (writer == null) { + return List.of(); + } + try { + return Arrays.asList(writer.complete().dataFiles()); + } finally { + writer = null; + resetBuffer(); + } + } + + /** Discard what has been written. Files already closed remain as orphans. */ + public void abort() { + if (writer == null) { + return; + } + try { + writer.abort(); + } catch (IOException e) { + LOG.warn("Failed aborting Iceberg writer; uncommitted files may remain as orphans", e); + } finally { + writer = null; + resetBuffer(); + } + } + + /** Roughly how many bytes the open files hold, for size-based flushing. */ + public long bufferedBytes() { + return countingAppenderFactory == null ? 0L : countingAppenderFactory.estimatedBytes(); + } + + /** + * Pick up schema and partition spec evolution. Without this the writer keeps using the + * metadata read at {@link #open()} until the worker restarts. + */ + public void refreshTable() { + table.refresh(); + } + + private void resetBuffer() { + if (countingAppenderFactory != null) { + countingAppenderFactory.reset(); + } + } + + private TaskWriter createWriter() { + Schema schema = table.schema(); + PartitionSpec spec = table.spec(); + FileFormat format = options.getFileFormat(); + long targetFileSize = options.getTargetFileSizeBytes() != null + ? options.getTargetFileSizeBytes() + : PropertyUtil.propertyAsLong(table.properties(), + TableProperties.WRITE_TARGET_FILE_SIZE_BYTES, + TableProperties.WRITE_TARGET_FILE_SIZE_BYTES_DEFAULT); + CountingAppenderFactory appenderFactory = + new CountingAppenderFactory(new GenericAppenderFactory(schema, spec)); + this.countingAppenderFactory = appenderFactory; + // A fresh OutputFileFactory per file set: its random operation id keeps file names from + // replayed tuples unique. + OutputFileFactory fileFactory = OutputFileFactory + .builderFor(table, taskId, System.currentTimeMillis()) + .format(format) + .build(); + if (spec.isUnpartitioned()) { + return new UnpartitionedWriter<>(spec, format, appenderFactory, fileFactory, table.io(), targetFileSize); + } + return new PartitionedRecordWriter(spec, format, appenderFactory, fileFactory, table.io(), targetFileSize, schema); + } + + @Override + public void close() { + abort(); + if (catalog instanceof Closeable) { + try { + ((Closeable) catalog).close(); + } catch (IOException e) { + LOG.warn("Failed closing Iceberg catalog", e); + } + } + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/PartitionedRecordWriter.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/PartitionedRecordWriter.java new file mode 100644 index 0000000000..56e1fda8c8 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/PartitionedRecordWriter.java @@ -0,0 +1,54 @@ +/* + * 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 org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionKey; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.InternalRecordWrapper; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.io.FileAppenderFactory; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.OutputFileFactory; +import org.apache.iceberg.io.PartitionedFanoutWriter; + +/** + * Fanout writer that routes generic {@link Record}s to one open file per table partition. + */ +class PartitionedRecordWriter extends PartitionedFanoutWriter { + + private final PartitionKey partitionKey; + private final InternalRecordWrapper wrapper; + + PartitionedRecordWriter(PartitionSpec spec, FileFormat format, FileAppenderFactory appenderFactory, + OutputFileFactory fileFactory, FileIO io, long targetFileSize, Schema schema) { + super(spec, format, appenderFactory, fileFactory, io, targetFileSize); + this.partitionKey = new PartitionKey(spec, schema); + this.wrapper = new InternalRecordWrapper(schema.asStruct()); + } + + @Override + protected PartitionKey partition(Record record) { + // PartitionedFanoutWriter copies the key when it caches a new partition's writer, + // so reusing one mutable key here is safe. + partitionKey.partition(wrapper.wrap(record)); + return partitionKey; + } +} diff --git a/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/RecordMapper.java b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/RecordMapper.java new file mode 100644 index 0000000000..c71d5ce3b5 --- /dev/null +++ b/external/storm-iceberg/src/main/java/org/apache/storm/iceberg/common/RecordMapper.java @@ -0,0 +1,40 @@ +/* + * 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.io.Serializable; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.Record; +import org.apache.storm.tuple.ITuple; + +/** + * Converts a {@link ITuple} into an Iceberg {@link Record} matching the target table schema. + * Implementations must be serializable: they are shipped with the topology. + */ +public interface RecordMapper extends Serializable { + + /** + * Convert a tuple to a record for the given table schema. + * + * @param tuple the input tuple + * @param schema the current schema of the target Iceberg table + * @return the record to write; never null + */ + Record map(ITuple tuple, Schema schema); +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/bolt/IcebergBoltTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/bolt/IcebergBoltTest.java new file mode 100644 index 0000000000..941bff32ee --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/bolt/IcebergBoltTest.java @@ -0,0 +1,218 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Types; +import org.apache.storm.Config; +import org.apache.storm.Constants; +import org.apache.storm.iceberg.common.CommitWal; +import org.apache.storm.iceberg.common.IcebergCommitter; +import org.apache.storm.iceberg.common.IcebergOptions; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.task.TopologyContext; +import org.apache.storm.tuple.Tuple; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class IcebergBoltTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get())); + private static final TableIdentifier TABLE_ID = TableIdentifier.of("db", "events"); + + @TempDir + Path tempDir; + + private String warehouse; + private HadoopCatalog verifyCatalog; + private OutputCollector collector; + + @BeforeEach + void setUp() { + warehouse = tempDir.toUri().toString(); + verifyCatalog = new HadoopCatalog(new Configuration(), warehouse); + verifyCatalog.createTable(TABLE_ID, SCHEMA, PartitionSpec.unpartitioned()); + collector = mock(OutputCollector.class); + } + + @AfterEach + void tearDown() throws IOException { + verifyCatalog.close(); + } + + private IcebergOptions.Builder baseOptions() { + Map catalogProps = new HashMap<>(); + catalogProps.put(CatalogUtil.ICEBERG_CATALOG_TYPE, CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP); + catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse); + return new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events"); + } + + private IcebergBolt prepared(IcebergOptions options) { + IcebergBolt bolt = new IcebergBolt(options); + Map topoConf = new HashMap<>(); + topoConf.put(Config.TOPOLOGY_NAME, "test-topology"); + TopologyContext context = mock(TopologyContext.class); + when(context.getThisTaskId()).thenReturn(7); + bolt.prepare(topoConf, context, collector); + return bolt; + } + + private Tuple tuple(long id, String name) { + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn("spout"); + when(tuple.contains(anyString())).thenReturn(false); + when(tuple.contains("id")).thenReturn(true); + when(tuple.contains("name")).thenReturn(true); + when(tuple.getValueByField("id")).thenReturn(id); + when(tuple.getValueByField("name")).thenReturn(name); + return tuple; + } + + private Tuple tickTuple() { + Tuple tuple = mock(Tuple.class); + when(tuple.getSourceComponent()).thenReturn(Constants.SYSTEM_COMPONENT_ID); + when(tuple.getSourceStreamId()).thenReturn(Constants.SYSTEM_TICK_STREAM_ID); + return tuple; + } + + private List readRows() { + Table table = verifyCatalog.loadTable(TABLE_ID); + table.refresh(); + List rows = new ArrayList<>(); + try (CloseableIterable records = IcebergGenerics.read(table).build()) { + records.forEach(rows::add); + } catch (IOException e) { + throw new RuntimeException(e); + } + return rows; + } + + @Test + void tuplesAreNotAckedBeforeTheirCommitLands() { + IcebergBolt bolt = prepared(baseOptions().withCommitIntervalRecords(3).build()); + + bolt.execute(tuple(1L, "alice")); + bolt.execute(tuple(2L, "bob")); + + verify(collector, never()).ack(org.mockito.ArgumentMatchers.any()); + assertEquals(0, readRows().size(), "nothing is visible before the commit"); + bolt.cleanup(); + } + + @Test + void reachingTheRecordThresholdCommitsAndAcksEveryBufferedTuple() { + IcebergBolt bolt = prepared(baseOptions().withCommitIntervalRecords(2).build()); + + bolt.execute(tuple(1L, "alice")); + bolt.execute(tuple(2L, "bob")); + + verify(collector, times(2)).ack(org.mockito.ArgumentMatchers.any()); + assertEquals(2, readRows().size()); + bolt.cleanup(); + } + + @Test + void aTickTupleFlushesWhatIsBuffered() { + IcebergBolt bolt = prepared(baseOptions().withCommitIntervalRecords(100).build()); + + bolt.execute(tuple(1L, "alice")); + bolt.execute(tickTuple()); + + verify(collector, times(1)).ack(org.mockito.ArgumentMatchers.any()); + assertEquals(1, readRows().size()); + bolt.cleanup(); + } + + @Test + void aTickTupleWithNothingBufferedCommitsNothing() { + IcebergBolt bolt = prepared(baseOptions().withCommitIntervalRecords(100).build()); + + bolt.execute(tickTuple()); + + verify(collector, never()).ack(org.mockito.ArgumentMatchers.any()); + assertEquals(0, readRows().size()); + bolt.cleanup(); + } + + @Test + void aFailedCommitFailsTheBufferedTuplesInsteadOfAckingThem() { + IcebergBolt bolt = prepared(baseOptions().withCommitIntervalRecords(2).build()); + bolt.execute(tuple(1L, "alice")); + // The table disappears underneath the bolt, so the commit cannot land. + verifyCatalog.dropTable(TABLE_ID, true); + + bolt.execute(tuple(2L, "bob")); + + verify(collector, never()).ack(org.mockito.ArgumentMatchers.any()); + verify(collector, times(2)).fail(org.mockito.ArgumentMatchers.any()); + } + + @Test + void prepareReplaysACommitLeftPendingByAnEarlierRun() { + Table table = verifyCatalog.loadTable(TABLE_ID); + CommitWal wal = new CommitWal(table, "test-topology", 7); + CommitWal.WalEntry pending = wal.write(List.of(DataFiles.builder(table.spec()) + .withPath(table.location() + "/data/left-behind.parquet") + .withFileSizeInBytes(1024L) + .withRecordCount(1L) + .withFormat(FileFormat.PARQUET) + .build())); + + IcebergBolt bolt = prepared(baseOptions().withCommitIntervalRecords(100).build()); + + table.refresh(); + assertEquals(pending.commitId(), + table.currentSnapshot().summary().get(IcebergCommitter.COMMIT_ID_PROPERTY), + "the pending commit is replayed on startup"); + assertEquals(List.of(), wal.listPending(), "and its WAL entry is cleared"); + bolt.cleanup(); + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/CommitWalTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/CommitWalTest.java new file mode 100644 index 0000000000..615ea63f16 --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/CommitWalTest.java @@ -0,0 +1,113 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class CommitWalTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get())); + private static final TableIdentifier TABLE_ID = TableIdentifier.of("db", "events"); + + @TempDir + Path tempDir; + + private HadoopCatalog catalog; + private Table table; + + @BeforeEach + void setUp() { + catalog = new HadoopCatalog(new Configuration(), tempDir.toUri().toString()); + table = catalog.createTable(TABLE_ID, SCHEMA, PartitionSpec.unpartitioned()); + } + + @AfterEach + void tearDown() throws IOException { + catalog.close(); + } + + private DataFile dataFile(String name, long records) { + return DataFiles.builder(table.spec()) + .withPath(table.location() + "/data/" + name) + .withFileSizeInBytes(1024L) + .withRecordCount(records) + .withFormat(FileFormat.PARQUET) + .build(); + } + + @Test + void aTaskThatHasNeverWrittenHasNothingPending() { + assertEquals(List.of(), new CommitWal(table, "topo", 0).listPending()); + } + + @Test + void aPendingEntryCarriesTheTimeItWasWritten() { + long before = System.currentTimeMillis(); + CommitWal wal = new CommitWal(table, "topo", 0); + + CommitWal.WalEntry written = wal.write(List.of(dataFile("a.parquet", 1L))); + long after = System.currentTimeMillis(); + + assertTrue(written.createdAtMs() >= before && written.createdAtMs() <= after, + "the entry records when it was written"); + assertEquals(written.createdAtMs(), wal.listPending().get(0).createdAtMs(), + "and listing recovers it without reading the entry"); + } + + @Test + void pendingEntryReadsBackTheDataFilesItWasWritten() { + CommitWal wal = new CommitWal(table, "topo", 3); + DataFile first = dataFile("a.parquet", 5L); + DataFile second = dataFile("b.parquet", 7L); + + CommitWal.WalEntry entry = wal.write(List.of(first, second)); + + List pending = wal.listPending(); + assertEquals(1, pending.size()); + assertEquals(entry.commitId(), pending.get(0).commitId()); + + List recovered = wal.read(pending.get(0)); + assertEquals(2, recovered.size()); + assertEquals(first.location(), recovered.get(0).location()); + assertEquals(5L, recovered.get(0).recordCount()); + assertEquals(second.location(), recovered.get(1).location()); + assertTrue(entry.location().contains("topo"), "WAL path should be scoped by topology"); + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/FieldNameRecordMapperTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/FieldNameRecordMapperTest.java new file mode 100644 index 0000000000..376f8368a7 --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/FieldNameRecordMapperTest.java @@ -0,0 +1,143 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.time.Instant; +import java.time.OffsetDateTime; +import java.time.ZoneOffset; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.types.Types; +import org.apache.storm.tuple.ITuple; +import org.junit.jupiter.api.Test; + +class FieldNameRecordMapperTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get()), + Types.NestedField.optional(3, "score", Types.DoubleType.get()), + Types.NestedField.optional(4, "ts", Types.TimestampType.withZone())); + + private final FieldNameRecordMapper mapper = new FieldNameRecordMapper(); + + private ITuple mockTuple() { + ITuple tuple = mock(ITuple.class); + when(tuple.contains(anyString())).thenReturn(false); + return tuple; + } + + private void field(ITuple tuple, String name, Object value) { + when(tuple.contains(name)).thenReturn(true); + when(tuple.getValueByField(name)).thenReturn(value); + } + + @Test + void mapsFieldsByName() { + ITuple tuple = mockTuple(); + field(tuple, "id", 42L); + field(tuple, "name", "storm"); + field(tuple, "score", 0.5d); + + Record record = mapper.map(tuple, SCHEMA); + + assertEquals(42L, record.getField("id")); + assertEquals("storm", record.getField("name")); + assertEquals(0.5d, record.getField("score")); + assertNull(record.getField("ts")); + } + + @Test + void widensNumericTypes() { + Schema schema = new Schema( + Types.NestedField.required(1, "i", Types.IntegerType.get()), + Types.NestedField.required(2, "l", Types.LongType.get()), + Types.NestedField.required(3, "f", Types.FloatType.get()), + Types.NestedField.required(4, "d", Types.DoubleType.get())); + ITuple tuple = mockTuple(); + field(tuple, "i", (short) 7); + field(tuple, "l", 7); + field(tuple, "f", 7); + field(tuple, "d", 7.0f); + + Record record = mapper.map(tuple, schema); + + assertEquals(7, record.getField("i")); + assertEquals(7L, record.getField("l")); + assertEquals(7.0f, record.getField("f")); + assertEquals((double) 7.0f, record.getField("d")); + } + + @Test + void convertsInstantAndEpochMillisToTimestamptz() { + Instant instant = Instant.parse("2026-07-12T10:15:30Z"); + ITuple tuple = mockTuple(); + field(tuple, "id", 1L); + field(tuple, "name", "x"); + field(tuple, "ts", instant); + + Record record = mapper.map(tuple, SCHEMA); + assertEquals(OffsetDateTime.ofInstant(instant, ZoneOffset.UTC), record.getField("ts")); + + ITuple tuple2 = mockTuple(); + field(tuple2, "id", 1L); + field(tuple2, "name", "x"); + field(tuple2, "ts", instant.toEpochMilli()); + + Record record2 = mapper.map(tuple2, SCHEMA); + assertEquals(OffsetDateTime.ofInstant(instant, ZoneOffset.UTC), record2.getField("ts")); + } + + @Test + void missingRequiredFieldThrows() { + ITuple tuple = mockTuple(); + field(tuple, "id", 1L); + // "name" (required) absent from the tuple + + IllegalArgumentException e = assertThrows(IllegalArgumentException.class, () -> mapper.map(tuple, SCHEMA)); + assertTrue(e.getMessage().contains("name")); + } + + @Test + void nullForRequiredFieldThrows() { + ITuple tuple = mockTuple(); + field(tuple, "id", 1L); + field(tuple, "name", null); + + assertThrows(IllegalArgumentException.class, () -> mapper.map(tuple, SCHEMA)); + } + + @Test + void missingOptionalFieldIsNull() { + ITuple tuple = mockTuple(); + field(tuple, "id", 1L); + field(tuple, "name", "x"); + + Record record = mapper.map(tuple, SCHEMA); + assertNull(record.getField("score")); + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergCommitterTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergCommitterTest.java new file mode 100644 index 0000000000..ffb3cfbe74 --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergCommitterTest.java @@ -0,0 +1,247 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.AppendFiles; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.CommitStateUnknownException; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class IcebergCommitterTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get())); + private static final TableIdentifier TABLE_ID = TableIdentifier.of("db", "events"); + + @TempDir + Path tempDir; + + private HadoopCatalog catalog; + private Table table; + private CommitWal wal; + private IcebergCommitter committer; + + @BeforeEach + void setUp() { + catalog = new HadoopCatalog(new Configuration(), tempDir.toUri().toString()); + table = catalog.createTable(TABLE_ID, SCHEMA, PartitionSpec.unpartitioned()); + wal = new CommitWal(table, "topo", 0); + committer = new IcebergCommitter(table, wal, new IcebergMetrics(null)); + } + + @AfterEach + void tearDown() throws IOException { + catalog.close(); + } + + private DataFile dataFile(String name) { + return DataFiles.builder(table.spec()) + .withPath(table.location() + "/data/" + name) + .withFileSizeInBytes(1024L) + .withRecordCount(4L) + .withFormat(FileFormat.PARQUET) + .build(); + } + + private List committedFileNames() { + table.refresh(); + Snapshot snapshot = table.currentSnapshot(); + if (snapshot == null) { + return List.of(); + } + List names = new ArrayList<>(); + for (DataFile file : snapshot.addedDataFiles(table.io())) { + names.add(file.location().substring(file.location().lastIndexOf('/') + 1)); + } + names.sort(String::compareTo); + return names; + } + + @Test + void commitMakesFilesVisibleAndClearsTheWal() { + committer.commit(List.of(dataFile("a.parquet"))); + + assertEquals(List.of("a.parquet"), committedFileNames()); + assertTrue(wal.listPending().isEmpty(), "a completed commit leaves no WAL entry"); + } + + @Test + void commitStampsItsCommitIdOnTheSnapshot() { + committer.commit(List.of(dataFile("a.parquet"))); + + table.refresh(); + String stamped = table.currentSnapshot().summary().get(IcebergCommitter.COMMIT_ID_PROPERTY); + assertTrue(stamped != null && !stamped.isBlank(), "snapshot should carry the commit id"); + } + + @Test + void theScanWindowStartsBeforeTheEntryByTheClockSkewAllowance() { + long entryCreatedAtMs = 1_000_000_000L; + long slack = IcebergCommitter.CLOCK_SKEW_ALLOWANCE_MS; + + // The snapshot that carries a commit is always written after its WAL entry, so anything + // older than the entry — beyond what clock skew between hosts can explain — cannot be it. + assertTrue(IcebergCommitter.withinScanWindow(entryCreatedAtMs + 1, entryCreatedAtMs)); + assertTrue(IcebergCommitter.withinScanWindow(entryCreatedAtMs - slack, entryCreatedAtMs)); + assertFalse(IcebergCommitter.withinScanWindow(entryCreatedAtMs - slack - 1, entryCreatedAtMs)); + } + + /** + * A committer whose append either throws before reaching the table, or reaches it and then + * throws as if the outcome were unknown. Everything else is the real table. + */ + private IcebergCommitter committerWithFlakyAppend(boolean landBeforeThrowing) { + return committerWithFlakyAppend(landBeforeThrowing, new RecordingMetricsContext()); + } + + private IcebergCommitter committerWithFlakyAppend(boolean landBeforeThrowing, + RecordingMetricsContext metrics) { + Table flakyTable = spy(table); + doAnswer(invocation -> { + AppendFiles real = table.newAppend(); + AppendFiles flaky = mock(AppendFiles.class); + when(flaky.set(anyString(), anyString())).thenAnswer(call -> { + real.set(call.getArgument(0), call.getArgument(1)); + return flaky; + }); + when(flaky.appendFile(any())).thenAnswer(call -> { + real.appendFile(call.getArgument(0)); + return flaky; + }); + doAnswer(commit -> { + if (landBeforeThrowing) { + real.commit(); + } + throw new CommitStateUnknownException(new RuntimeException("commit outcome unknown")); + }).when(flaky).commit(); + return flaky; + }).when(flakyTable).newAppend(); + return new IcebergCommitter(flakyTable, wal, new IcebergMetrics(metrics)); + } + + @Test + void aCommitThatLandedDespiteAFailureIsCountedAsCommittedNotFailed() { + RecordingMetricsContext metrics = new RecordingMetricsContext(); + + committerWithFlakyAppend(true, metrics).commit(List.of(dataFile("a.parquet"))); + + assertEquals(1L, metrics.counter(IcebergMetrics.DATA_FILES_COMMITTED), + "the file is visible, so it counts as committed"); + assertEquals(1024L, metrics.counter(IcebergMetrics.BYTES_COMMITTED)); + assertEquals(1L, metrics.timerCount(IcebergMetrics.COMMIT_LATENCY)); + assertEquals(0L, metrics.counter(IcebergMetrics.COMMIT_FAILURES), + "a commit that landed is not a failure"); + } + + @Test + void aCommitThatDidNotLandIsCountedAsFailed() { + RecordingMetricsContext metrics = new RecordingMetricsContext(); + IcebergCommitter flaky = committerWithFlakyAppend(false, metrics); + + assertThrows(CommitStateUnknownException.class, + () -> flaky.commit(List.of(dataFile("a.parquet")))); + + assertEquals(1L, metrics.counter(IcebergMetrics.COMMIT_FAILURES)); + assertEquals(0L, metrics.counter(IcebergMetrics.DATA_FILES_COMMITTED), + "nothing became visible, so nothing counts as committed"); + } + + @Test + void aCommitThatLandedDespiteAFailureIsTreatedAsSuccessful() { + // CommitStateUnknownException where the commit did reach the table: the snapshot carries + // the commit id, which settles the question, so there is nothing to fail or replay. + committerWithFlakyAppend(true).commit(List.of(dataFile("a.parquet"))); + + assertEquals(List.of("a.parquet"), committedFileNames()); + assertTrue(wal.listPending().isEmpty(), "the settled entry is cleared in flight"); + } + + @Test + void aCommitThatDidNotLandClearsItsWalEntryBeforeFailing() { + // The tuples are about to be failed and replayed. Leaving the entry behind would make the + // next startup append these files too, duplicating what the replay writes. + IcebergCommitter flaky = committerWithFlakyAppend(false); + + assertThrows(CommitStateUnknownException.class, + () -> flaky.commit(List.of(dataFile("a.parquet")))); + + assertEquals(List.of(), committedFileNames(), "nothing became visible"); + assertTrue(wal.listPending().isEmpty(), "and no entry is left for startup to replay"); + } + + @Test + void recoveryReplaysAPreparedCommitThatNeverBecameVisible() { + // A crash between "data files durable, WAL written" and "Iceberg commit": the entry exists, + // no snapshot references it. + wal.write(List.of(dataFile("lost.parquet"))); + assertEquals(List.of(), committedFileNames()); + + int replayed = committer.recover(); + + assertEquals(1, replayed); + assertEquals(List.of("lost.parquet"), committedFileNames()); + assertTrue(wal.listPending().isEmpty(), "the replayed entry is cleared"); + } + + @Test + void recoveryDoesNotReappendACommitThatIsAlreadyVisible() { + // A crash between the Iceberg commit and the WAL delete: the snapshot is there, so the + // entry must be dropped rather than replayed, or the batch would be committed twice. + CommitWal.WalEntry entry = wal.write(List.of(dataFile("a.parquet"))); + table.newAppend() + .appendFile(dataFile("a.parquet")) + .set(IcebergCommitter.COMMIT_ID_PROPERTY, entry.commitId()) + .commit(); + + int replayed = committer.recover(); + + assertEquals(0, replayed); + assertEquals(List.of("a.parquet"), committedFileNames()); + assertTrue(wal.listPending().isEmpty(), "the settled entry is cleared"); + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergOptionsTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergOptionsTest.java new file mode 100644 index 0000000000..fbbc4e831e --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergOptionsTest.java @@ -0,0 +1,88 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectOutputStream; +import java.util.Map; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; + +class IcebergOptionsTest { + + private static final Map CATALOG_PROPS = Map.of("type", "hadoop", "warehouse", "file:///tmp/wh"); + + private IcebergOptions.Builder validBuilder() { + return new IcebergOptions.Builder() + .withCatalogProperties(CATALOG_PROPS) + .withTable("db.events"); + } + + @Test + void buildsWithDefaults() { + IcebergOptions options = validBuilder().build(); + + assertEquals(CATALOG_PROPS, options.getCatalogProperties()); + assertEquals("db.events", options.getTableIdentifier()); + assertInstanceOf(FieldNameRecordMapper.class, options.getRecordMapper()); + assertEquals(FileFormat.PARQUET, options.getFileFormat()); + assertNull(options.getTargetFileSizeBytes()); + assertNull(options.getAutoCreateSchema()); + assertNull(options.getAutoCreateSpec()); + } + + @Test + void rejectsMissingCatalogProperties() { + IcebergOptions.Builder builder = new IcebergOptions.Builder().withTable("db.events"); + assertThrows(IllegalStateException.class, builder::build); + } + + @Test + void rejectsMissingTable() { + IcebergOptions.Builder builder = new IcebergOptions.Builder().withCatalogProperties(CATALOG_PROPS); + assertThrows(IllegalStateException.class, builder::build); + } + + @Test + void rejectsNonPositiveTargetFileSize() { + assertThrows(IllegalStateException.class, () -> validBuilder().withTargetFileSizeBytes(0).build()); + } + + @Test + void isJavaSerializableWithAutoCreate() throws IOException { + Schema schema = new Schema(Types.NestedField.required(1, "id", Types.LongType.get())); + IcebergOptions options = validBuilder() + .withAutoCreate(schema, PartitionSpec.unpartitioned()) + .withTargetFileSizeBytes(1024L) + .build(); + + try (ObjectOutputStream out = new ObjectOutputStream(new ByteArrayOutputStream())) { + out.writeObject(options); // must not throw NotSerializableException + } + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergWriterTest.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergWriterTest.java new file mode 100644 index 0000000000..38e9766e7c --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/IcebergWriterTest.java @@ -0,0 +1,170 @@ +/* + * 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 static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.CatalogProperties; +import org.apache.iceberg.CatalogUtil; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.data.IcebergGenerics; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.types.Types; +import org.apache.storm.tuple.ITuple; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class IcebergWriterTest { + + private static final Schema SCHEMA = new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.required(2, "name", Types.StringType.get())); + private static final TableIdentifier TABLE_ID = TableIdentifier.of("db", "events"); + + @TempDir + Path tempDir; + + private String warehouse; + private HadoopCatalog verifyCatalog; + + @BeforeEach + void setUp() { + warehouse = tempDir.toUri().toString(); + verifyCatalog = new HadoopCatalog(new Configuration(), warehouse); + } + + @AfterEach + void tearDown() throws IOException { + verifyCatalog.close(); + } + + private IcebergOptions.Builder baseOptions() { + Map catalogProps = new HashMap<>(); + catalogProps.put(CatalogUtil.ICEBERG_CATALOG_TYPE, CatalogUtil.ICEBERG_CATALOG_TYPE_HADOOP); + catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse); + return new IcebergOptions.Builder() + .withCatalogProperties(catalogProps) + .withTable("db.events"); + } + + private ITuple tuple(long id, String name) { + ITuple tuple = mock(ITuple.class); + when(tuple.contains(anyString())).thenReturn(false); + when(tuple.contains("id")).thenReturn(true); + when(tuple.contains("name")).thenReturn(true); + when(tuple.getValueByField("id")).thenReturn(id); + when(tuple.getValueByField("name")).thenReturn(name); + return tuple; + } + + private List readRows() { + Table table = verifyCatalog.loadTable(TABLE_ID); + table.refresh(); + List rows = new ArrayList<>(); + try (CloseableIterable records = IcebergGenerics.read(table).build()) { + records.forEach(rows::add); + } catch (IOException e) { + throw new RuntimeException(e); + } + return rows; + } + + @Test + void writtenTuplesBecomeReadableRowsOnceCommitted() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA, PartitionSpec.unpartitioned()); + try (IcebergWriter writer = new IcebergWriter(baseOptions().build(), 0)) { + writer.open(); + writer.write(tuple(1L, "alice")); + writer.write(tuple(2L, "bob")); + + List dataFiles = writer.complete(); + assertFalse(dataFiles.isEmpty(), "completing a non-empty writer yields data files"); + + CommitWal wal = new CommitWal(writer.table(), "topo", 0); + new IcebergCommitter(writer.table(), wal, new IcebergMetrics(null)).commit(dataFiles); + } + + assertEquals(2, readRows().size()); + } + + @Test + void tableIsCreatedOnFirstUseWhenAutoCreateIsConfigured() throws IOException { + try (IcebergWriter writer = new IcebergWriter( + baseOptions().withAutoCreate(SCHEMA, PartitionSpec.unpartitioned()).build(), 0)) { + writer.open(); + assertTrue(verifyCatalog.tableExists(TABLE_ID)); + } + } + + @Test + void completingAnEmptyWriterYieldsNoDataFiles() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA, PartitionSpec.unpartitioned()); + try (IcebergWriter writer = new IcebergWriter(baseOptions().build(), 0)) { + writer.open(); + assertEquals(List.of(), writer.complete()); + } + } + + @Test + void partitionedTablesFanOutOneFilePerPartition() throws IOException { + PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).identity("name").build(); + verifyCatalog.createTable(TABLE_ID, SCHEMA, spec); + try (IcebergWriter writer = new IcebergWriter(baseOptions().build(), 0)) { + writer.open(); + writer.write(tuple(1L, "alice")); + writer.write(tuple(2L, "bob")); + + assertEquals(2, writer.complete().size(), "one data file per partition value"); + } + } + + @Test + void bufferedBytesGrowWithWritesAndResetOnComplete() throws IOException { + verifyCatalog.createTable(TABLE_ID, SCHEMA, PartitionSpec.unpartitioned()); + try (IcebergWriter writer = new IcebergWriter(baseOptions().build(), 0)) { + writer.open(); + assertEquals(0L, writer.bufferedBytes()); + writer.write(tuple(1L, "alice")); + assertTrue(writer.bufferedBytes() > 0L, "a written tuple counts towards the buffer"); + + writer.complete(); + assertEquals(0L, writer.bufferedBytes(), "completing starts a fresh buffer"); + } + } +} diff --git a/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/RecordingMetricsContext.java b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/RecordingMetricsContext.java new file mode 100644 index 0000000000..9eedf77f1f --- /dev/null +++ b/external/storm-iceberg/src/test/java/org/apache/storm/iceberg/common/RecordingMetricsContext.java @@ -0,0 +1,102 @@ +/* + * 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.codahale.metrics.Counter; +import com.codahale.metrics.Gauge; +import com.codahale.metrics.Histogram; +import com.codahale.metrics.Meter; +import com.codahale.metrics.MetricSet; +import com.codahale.metrics.Timer; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.metric.api.CombinedMetric; +import org.apache.storm.metric.api.ICombiner; +import org.apache.storm.metric.api.IMetric; +import org.apache.storm.metric.api.IReducer; +import org.apache.storm.metric.api.ReducedMetric; +import org.apache.storm.task.IMetricsContext; + +/** + * Minimal {@link IMetricsContext} that keeps the registered metrics in maps so tests can assert on + * them. Like the real registry, registering the same name twice returns the same instance. + */ +class RecordingMetricsContext implements IMetricsContext { + + private final Map counters = new HashMap<>(); + private final Map timers = new HashMap<>(); + + long counter(String name) { + Counter counter = counters.get(name); + return counter == null ? 0L : counter.getCount(); + } + + long timerCount(String name) { + Timer timer = timers.get(name); + return timer == null ? 0L : timer.getCount(); + } + + @Override + public Counter registerCounter(String name) { + return counters.computeIfAbsent(name, n -> new Counter()); + } + + @Override + public Timer registerTimer(String name) { + return timers.computeIfAbsent(name, n -> new Timer()); + } + + @Override + public Histogram registerHistogram(String name) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + public Meter registerMeter(String name) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + public Gauge registerGauge(String name, Gauge gauge) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + public void registerMetricSet(String prefix, MetricSet set) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + @Deprecated + public T registerMetric(String name, T metric, int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + @Deprecated + public ReducedMetric registerMetric(String name, IReducer reducer, int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("not used by IcebergState"); + } + + @Override + @Deprecated + public CombinedMetric registerMetric(String name, ICombiner combiner, int timeBucketSizeInSecs) { + throw new UnsupportedOperationException("not used by IcebergState"); + } +} diff --git a/pom.xml b/pom.xml index ed266c241f..8764757bf3 100644 --- a/pom.xml +++ b/pom.xml @@ -113,6 +113,9 @@ 7.1.0 3.5.0 2.6.6-hadoop3 + + 1.11.0 5.6.2 3.5 6.1.0 @@ -896,6 +899,35 @@ bcprov-jdk18on ${bouncycastle.version} + + + org.apache.iceberg + iceberg-api + ${iceberg.version} + + + org.apache.iceberg + iceberg-core + ${iceberg.version} + + + org.apache.iceberg + iceberg-data + ${iceberg.version} + + + org.apache.iceberg + iceberg-parquet + ${iceberg.version} + + + org.apache.iceberg + iceberg-orc + ${iceberg.version} + org.apache.hadoop diff --git a/storm-dist/binary/final-package/src/main/assembly/common.xml b/storm-dist/binary/final-package/src/main/assembly/common.xml index 402fd3837b..999a9a798b 100644 --- a/storm-dist/binary/final-package/src/main/assembly/common.xml +++ b/storm-dist/binary/final-package/src/main/assembly/common.xml @@ -133,6 +133,13 @@ README.* + + ${project.basedir}/../../../external/storm-iceberg + external/storm-iceberg + + README.* + + ${project.basedir}/../../../external/storm-eventhubs external/storm-eventhubs