Class ColumnarRunner

java.lang.Object
io.krabka.streams.columnar.ColumnarRunner
public final class ColumnarRunner extends Object
Runs fetch, process, asynchronously produce, and commit cycles.

The runner is the bridge between plain Kafka clients and a columnar topology. Each cycle polls the consumer, evaluates the topology per logical partition, produces every output record concurrently, and commits offsets only after all acknowledgements arrive — so a crash mid-cycle reprocesses rather than loses records. On failure the affected partition's operator state is rolled back to its pre-batch snapshot before the error policy is applied.

Three styles are available: the static runPartitionOnce methods for a manually assigned partition, the static runGroupOnce methods for a subscribed consumer group, and group, which returns a ColumnarRunner.GroupRunner that also manages rebalances and persistent operator state. All methods drive the callers' consumer and producer; the runner creates no clients and owns no threads, so the caller controls the loop, threading, and shutdown.

Example

try (var runner = ColumnarRunner.group(topology, consumer, producer,
        ColumnarErrorPolicy.deadLetter("transactions-dead-letter"),
        new FileColumnarStateStore(stateDirectory),
        new ColumnarMetrics())) {
    while (running) {
        runner.runOnce(Duration.ofSeconds(1));
    }
}

  • Nested Class Summary

    Nested Classes
    Modifier and Type
    Class
    Description
    static final class 
    ColumnarRunner.GroupRunner
    A consumer-group runner that manages rebalances and persistent operator state.

  • Method Summary

    Modifier and Type
    Method
    Description
    static ColumnarRunner.GroupRunner
    group(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer)
    Creates an ephemeral, fail-fast group runner.
    static ColumnarRunner.GroupRunner
    group(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, ColumnarErrorPolicy errorPolicy, ColumnarStateStore stateStore, ColumnarMetrics metrics)
    Creates a group runner with explicit failure, snapshot, and metric handling.
    static ColumnarRunner.GroupRunner
    group(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, ColumnarErrorPolicy errorPolicy, ColumnarStateStore stateStore, ColumnarMetrics metrics, BarrierAlignment barrier)
    Creates a group runner that aligns its work on a barrier group's cuts.
    static Map<org.apache.kafka.common.TopicPartition, Long>
    runGroupOnce(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, Duration pollTimeout)
    Polls and processes every partition assigned by the consumer group, then commits their offsets.
    static Map<org.apache.kafka.common.TopicPartition, Long>
    runGroupOnce(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, Duration pollTimeout, ColumnarErrorPolicy errorPolicy, ColumnarMetrics metrics)
    Polls and processes every assigned partition with explicit failure handling and metrics.
    static Map<org.apache.kafka.common.TopicPartition, Long>
    runGroupOnceTransactional(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, Duration pollTimeout)
    Atomically produces a group poll and commits its consumed offsets in the producer transaction.
    static Map<org.apache.kafka.common.TopicPartition, Long>
    runGroupOnceTransactional(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, Duration pollTimeout, ColumnarErrorPolicy errorPolicy, ColumnarMetrics metrics)
    Runs one exactly-once cycle: output records and consumed offsets commit in one producer transaction.
    static long
    runPartitionOnce(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, String topic, int partition, long offset, Duration pollTimeout)
    Runs one fetch, process, produce, and commit cycle for one manually assigned partition.
    static long
    runPartitionOnce(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, String topic, int partition, long offset, Duration pollTimeout)
    Builds the topology, runs one cycle for one partition, and discards the state.
    static CompletableFuture<Void>
    sendAsync(List<ProducedToTopic> outputs, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer)
    Sends all records concurrently and completes when every broker acknowledgement arrives.
    static void
    subscribe(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer)
    Subscribes a group-managed consumer to every source topic in the topology.

    Methods inherited from class Object

    clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Method Details

    • runPartitionOnce

      public static long runPartitionOnce(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, String topic, int partition, long offset, Duration pollTimeout)
      Builds the topology, runs one cycle for one partition, and discards the state.

      Because the built topology is thrown away, stateful operators start fresh every call; use the BuiltColumnarTopology overload to accumulate state across cycles.

      Parameters:
      topology - the topology to build and run once
      consumer - the consumer to assign and poll; its subscription is replaced
      producer - the producer output records are sent with
      topic - the source topic to read
      partition - the partition to read
      offset - the offset to seek to before polling
      pollTimeout - how long one poll may block
      Returns:
      the next offset to process from; offset itself when the poll was empty

    • runPartitionOnce

      public static long runPartitionOnce(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, String topic, int partition, long offset, Duration pollTimeout)
      Runs one fetch, process, produce, and commit cycle for one manually assigned partition.

      The consumer is assigned to the partition and sought to the offset. When the poll returns records, the batch is processed, every output is produced and acknowledged, and the next offset is committed synchronously. If any step fails, the partition's operator state is rolled back to its pre-batch snapshot and the error is rethrown, so the caller can retry from the same offset.

      Parameters:
      topology - the built topology whose partition state carries across calls
      consumer - the consumer to assign and poll; its subscription is replaced
      producer - the producer output records are sent with
      topic - the source topic to read
      partition - the partition to read
      offset - the offset to seek to before polling
      pollTimeout - how long one poll may block
      Returns:
      the next offset to process from; offset itself when the poll was empty

    • runGroupOnce

      public static Map<org.apache.kafka.common.TopicPartition, Long> runGroupOnce(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, Duration pollTimeout)
      Polls and processes every partition assigned by the consumer group, then commits their offsets.

      Fail-fast form: any processing failure rolls the affected partitions back and rethrows. Equivalent to the six-argument overload with ColumnarErrorPolicy.fail() and throwaway metrics.

      Parameters:
      topology - the built topology whose partition state carries across calls
      consumer - the subscribed consumer to poll
      producer - the producer output records are sent with
      pollTimeout - how long one poll may block
      Returns:
      topic partition to next offset for everything committed this cycle; empty when the poll returned nothing

    • runGroupOnce

      public static Map<org.apache.kafka.common.TopicPartition, Long> runGroupOnce(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, Duration pollTimeout, ColumnarErrorPolicy errorPolicy, ColumnarMetrics metrics)
      Polls and processes every assigned partition with explicit failure handling and metrics.

      Each logical partition is processed independently: a partition that fails is rolled back and handled per the error policy, while the others' output is still produced and committed. If producing or committing fails, every partition of the cycle is rolled back and the error is rethrown.

      Parameters:
      topology - the built topology whose partition state carries across calls
      consumer - the subscribed consumer to poll
      producer - the producer output records are sent with
      pollTimeout - how long one poll may block
      errorPolicy - what to do with a partition whose processing fails
      metrics - the counters one observation per partition batch is recorded to
      Returns:
      topic partition to next offset for everything committed this cycle; empty when the poll returned nothing

    • runGroupOnceTransactional

      public static Map<org.apache.kafka.common.TopicPartition, Long> runGroupOnceTransactional(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, Duration pollTimeout)
      Atomically produces a group poll and commits its consumed offsets in the producer transaction.

      Fail-fast form of the transactional cycle; see the six-argument overload. The producer must be transactional and already initialized with initTransactions().

      Parameters:
      topology - the built topology whose partition state carries across calls
      consumer - the subscribed consumer to poll
      producer - the transactional producer output records are sent with
      pollTimeout - how long one poll may block
      Returns:
      topic partition to next offset for everything committed this cycle; empty when the poll returned nothing

    • runGroupOnceTransactional

      public static Map<org.apache.kafka.common.TopicPartition, Long> runGroupOnceTransactional(BuiltColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, Duration pollTimeout, ColumnarErrorPolicy errorPolicy, ColumnarMetrics metrics)
      Runs one exactly-once cycle: output records and consumed offsets commit in one producer transaction.

      The whole poll — output records and offset advances — becomes visible atomically or not at all. On any failure the transaction is aborted, every partition of the cycle is rolled back, and the error is rethrown. The producer must be transactional and already initialized with initTransactions(); downstream consumers need isolation.level=read_committed to observe exactly-once behavior.

      Parameters:
      topology - the built topology whose partition state carries across calls
      consumer - the subscribed consumer to poll
      producer - the transactional producer output records are sent with
      pollTimeout - how long one poll may block
      errorPolicy - what to do with a partition whose processing fails
      metrics - the counters one observation per partition batch is recorded to
      Returns:
      topic partition to next offset for everything committed this cycle; empty when the poll returned nothing

    • sendAsync

      public static CompletableFuture<Void> sendAsync(List<ProducedToTopic> outputs, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer)
      Sends all records concurrently and completes when every broker acknowledgement arrives.

      Records enter the producer in list order but acknowledgements may arrive in any order. A record whose timestamp is negative is sent without a timestamp, so the producer or broker assigns one. The returned future fails when any send fails; the remaining sends are not canceled.

      Parameters:
      outputs - the records to send, each bound to its topic
      producer - the producer to send with
      Returns:
      a future that completes when every record is acknowledged

    • subscribe

      public static void subscribe(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer)
      Subscribes a group-managed consumer to every source topic in the topology.

      Use this together with the static runGroupOnce methods; group subscribes its consumer itself.

      Parameters:
      topology - the topology whose source topics are subscribed to
      consumer - the group-managed consumer to subscribe

    • group

      public static ColumnarRunner.GroupRunner group(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer)
      Creates an ephemeral, fail-fast group runner.

      Equivalent to the six-argument overload with ColumnarErrorPolicy.fail(), ColumnarStateStore.none(), and fresh metrics: processing failures are rethrown and operator state does not survive rebalances.

      Parameters:
      topology - the topology to build and run; the runner owns the built form
      consumer - the consumer the runner polls; it is subscribed to the topology's source topics
      producer - the producer output records are sent with
      Returns:
      the runner; close it to save nothing and release processor state

    • group

      public static ColumnarRunner.GroupRunner group(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, ColumnarErrorPolicy errorPolicy, ColumnarStateStore stateStore, ColumnarMetrics metrics)
      Creates a group runner with explicit failure, snapshot, and metric handling.

      The runner builds the topology, subscribes the consumer to its source topics with itself as the rebalance listener, and from then on loads operator state from the state store when a logical partition is assigned and saves it back when the partition is revoked or the runner closes. Lost partitions are released without saving.

      Parameters:
      topology - the topology to build and run; the runner owns the built form
      consumer - the consumer the runner polls; it is subscribed to the topology's source topics
      producer - the producer output records are sent with
      errorPolicy - what to do with a partition whose processing fails
      stateStore - where operator snapshots are loaded from and saved to
      metrics - the counters the runner records observations to
      Returns:
      the runner; close it to save state and release the built topology

    • group

      public static ColumnarRunner.GroupRunner group(ColumnarTopology topology, org.apache.kafka.clients.consumer.Consumer<byte[],byte[]> consumer, org.apache.kafka.clients.producer.Producer<byte[],byte[]> producer, ColumnarErrorPolicy errorPolicy, ColumnarStateStore stateStore, ColumnarMetrics metrics, BarrierAlignment barrier)
      Creates a group runner that aligns its work on a barrier group's cuts.

      Behaves as the six-argument overload, and additionally reads the barrier group's cuts from __barrier_state. Records at or above a partition's marker offset wait for the next cycle, and the partition pauses until every assigned partition reaches the cut. The runner then snapshots each owned partition under the cut's epoch, commits the cut offsets, and calls the alignment's listener. ColumnarRunner.GroupRunner.restoreToEpoch(long) and ColumnarRunner.GroupRunner.restoreToLatestCut() read such a snapshot back and seek every input to the cut.

      Parameters:
      topology - the topology to build and run; the runner owns the built form
      consumer - the consumer the runner polls; it is subscribed to the topology's source topics
      producer - the producer output records are sent with
      errorPolicy - what to do with a partition whose processing fails
      stateStore - where operator snapshots are loaded from and saved to
      metrics - the counters the runner records observations to
      barrier - the barrier group to align on, or null for no alignment
      Returns:
      the runner; close it to save state and release the built topology