Class BuiltColumnarTopology

java.lang.Object
io.krabka.streams.columnar.BuiltColumnarTopology
All Implemented Interfaces:
AutoCloseable
public final class BuiltColumnarTopology extends Object implements AutoCloseable
A validated and reusable columnar topology.

Created by ColumnarTopology.build(). The built form owns the processor instances, one set per logical partition number, created lazily when a partition first processes records. Stateful operators therefore accumulate across calls: keep one built topology for the lifetime of a running partition or consumer group member, and close() it to release processor state.

All methods are synchronized; one built topology processes one batch at a time. Every intermediate Arrow batch created during an evaluation is closed before the method returns, so callers only handle the returned byte-backed records.

Example

try (var built = topology.build()) {
    List<ProducedToTopic> output = built.runBatch("transactions", records);

    Map<String, byte[]> snapshots = built.snapshotPartition(0);
    built.releasePartition(0);
    built.restorePartition(0, snapshots);
}

  • Method Details

    • runBatch

      public List<ProducedToTopic> runBatch(String topic, List<ConsumedRecord> records)
      Runs one source topic's records through the graph.

      Records are grouped by their partition number and each partition is evaluated against its own processor state, in ascending partition order.

      Parameters:
      topic - the source topic the records belong to
      records - the fetched records
      Returns:
      everything the topology's sinks produced, in evaluation order

    • runBatches

      public List<ProducedToTopic> runBatches(Map<String, List<ConsumedRecord>> input)
      Runs records from multiple source topics as one graph evaluation, enabling fan-in.

      Use this form when a join or merge needs both sides' records in the same evaluation. Records are grouped by partition number across all topics, so co-partitioned topics join partition by partition.

      Parameters:
      input - source topic to fetched records
      Returns:
      everything the topology's sinks produced, in evaluation order

    • runPartitionBatches

      public List<ProducedToTopic> runPartitionBatches(int partition, Map<String, List<ConsumedRecord>> input)
      Evaluates one co-partitioned set of source batches against isolated processor state.

      This is the lowest-level entry point, used by the runners after they have grouped a poll by partition. All records must carry the given partition number. If the evaluation throws, partially forwarded batches are closed, but operator state may have advanced; runners roll the partition back with restorePartition(int, Map) in that case.

      Parameters:
      partition - the logical partition whose processor state is used
      input - source topic to that partition's fetched records
      Returns:
      everything the topology's sinks produced, in evaluation order
      Throws:
      IllegalArgumentException - if a record belongs to another partition
      ColumnarException - if decoding, an operator, or encoding fails

    • snapshotPartition

      public Map<String,byte[]> snapshotPartition(int partition)
      Returns snapshots keyed by operator name for one logical partition.

      Only StatefulColumnarProcessor nodes, including joins, appear in the result. A partition that has never processed records returns an empty map.

      Parameters:
      partition - the logical partition to snapshot
      Returns:
      operator name to snapshot bytes; empty when the partition has no state

    • restorePartition

      public void restorePartition(int partition, Map<String,byte[]> snapshots)
      Restores snapshots before processing the partition. Unknown operator names are ignored.

      Restoring creates the partition's processors if needed and replaces the state of every stateful operator whose name appears in the map. Snapshots are matched by node name, which is why names should stay stable across application versions.

      Parameters:
      partition - the logical partition to restore
      snapshots - operator name to snapshot bytes, as returned by snapshotPartition(int)
      Throws:
      ColumnarException - if a named operator cannot restore its bytes

    • releasePartition

      public void releasePartition(int partition)
      Drops all processor state owned by a logical partition.

      Closeable processors are closed. The next batch for the partition starts from fresh processor instances.

      Parameters:
      partition - the logical partition to release
      Throws:
      ColumnarException - if a processor fails to close

    • close

      public void close()
      Releases every partition's processor state.

      The underlying ColumnarTopology and allocator are not touched and can build again.

      Specified by:
      close in interface AutoCloseable
      Throws:
      ColumnarException - if a processor fails to close