Class ColumnarRunner.GroupRunner

java.lang.Object
io.krabka.streams.columnar.ColumnarRunner.GroupRunner
All Implemented Interfaces:
AutoCloseable, org.apache.kafka.clients.consumer.ConsumerRebalanceListener
Enclosing class:
ColumnarRunner
public static final class ColumnarRunner.GroupRunner extends Object implements org.apache.kafka.clients.consumer.ConsumerRebalanceListener, AutoCloseable
A consumer-group runner that manages rebalances and persistent operator state.

Created by ColumnarRunner.group(ColumnarTopology, Consumer, Producer). As the consumer's rebalance listener it restores each newly assigned logical partition from the state store and saves each revoked partition back, so stateful operators survive rebalances and clean restarts. The caller drives processing by calling runOnce(Duration) or runOnceTransactional(Duration) in a loop; the runner owns no threads.

A runner built with a BarrierAlignment also aligns on the cuts of a barrier group. It holds back every record at or above a partition's marker offset, pauses that partition, and fires the barrier once every assigned partition reaches the cut. Firing snapshots each owned partition under the cut's epoch and commits the cut offsets, so the stored state is exactly the committed state at the cut.

Closing the runner saves and releases every partition it still owns and closes the built topology. The consumer and producer remain the caller's to close.

  • Method Details

    • runOnce

      public Map<org.apache.kafka.common.TopicPartition, Long> runOnce(Duration pollTimeout)
      Runs one fetch, process, produce, and commit cycle.

      See ColumnarRunner.runGroupOnce(BuiltColumnarTopology, Consumer, Producer, Duration, ColumnarErrorPolicy, ColumnarMetrics) for the cycle's semantics; this method uses the runner's own topology, clients, policy, and metrics.

      Parameters:
      pollTimeout - how long one poll may block
      Returns:
      topic partition to next offset for everything committed this cycle; empty when the poll returned nothing

    • runOnceTransactional

      public Map<org.apache.kafka.common.TopicPartition, Long> runOnceTransactional(Duration pollTimeout)
      Runs one exactly-once cycle in a producer transaction.

      See ColumnarRunner.runGroupOnceTransactional(BuiltColumnarTopology, Consumer, Producer, Duration, ColumnarErrorPolicy, ColumnarMetrics) for the cycle's semantics; the runner's producer must be transactional and already initialized with initTransactions().

      Parameters:
      pollTimeout - how long one poll may block
      Returns:
      topic partition to next offset for everything committed this cycle; empty when the poll returned nothing

    • metrics

      public ColumnarMetrics metrics()
      Returns the counters this runner records observations to.
      Returns:
      the metrics supplied when the runner was created

    • pendingCut

      public Optional<BarrierCut> pendingCut()
      Returns the cut this runner is waiting for.
      Returns:
      the pending cut, or empty when no barrier is pending

    • restoreToEpoch

      public BarrierCut restoreToEpoch(long epoch)
      Restores every owned partition to a barrier epoch and seeks the inputs to the cut.

      The runner drops the current operator state of each owned logical partition, loads the epoch's snapshot from the state store, and seeks every assigned partition that the cut names to its marker offset. Processing then resumes exactly at the cut. Call it while the runner is not inside runOnce.

      Parameters:
      epoch - the barrier epoch to restore
      Returns:
      the cut the runner restored to
      Throws:
      ColumnarException - if the runner has no barrier alignment or the group has no complete cut for the epoch
      IllegalArgumentException - if the epoch is negative

    • restoreToLatestCut

      public Optional<BarrierCut> restoreToLatestCut()
      Restores every owned partition to the newest complete cut of the group.
      Returns:
      the cut the runner restored to, or empty when the group has none
      Throws:
      ColumnarException - if the runner has no barrier alignment

    • onPartitionsAssigned

      public void onPartitionsAssigned(Collection<org.apache.kafka.common.TopicPartition> partitions)
      Restores newly assigned logical partitions from the state store.

      Invoked by the consumer during rebalancing; do not call directly.

      Specified by:
      onPartitionsAssigned in interface org.apache.kafka.clients.consumer.ConsumerRebalanceListener
      Parameters:
      partitions - the topic partitions the consumer was assigned

    • onPartitionsRevoked

      public void onPartitionsRevoked(Collection<org.apache.kafka.common.TopicPartition> partitions)
      Saves and releases logical partitions the consumer is giving up.

      Invoked by the consumer during rebalancing; do not call directly.

      Specified by:
      onPartitionsRevoked in interface org.apache.kafka.clients.consumer.ConsumerRebalanceListener
      Parameters:
      partitions - the topic partitions the consumer is revoking

    • onPartitionsLost

      public void onPartitionsLost(Collection<org.apache.kafka.common.TopicPartition> partitions)
      Releases lost logical partitions without saving their state.

      Lost partitions may already be owned elsewhere, so their possibly stale state is discarded rather than saved. Invoked by the consumer; do not call directly.

      Specified by:
      onPartitionsLost in interface org.apache.kafka.clients.consumer.ConsumerRebalanceListener
      Parameters:
      partitions - the topic partitions the consumer lost

    • close

      public void close()
      Saves every still-owned partition's state and closes the built topology.

      The consumer and producer are not closed; they belong to the caller.

      Specified by:
      close in interface AutoCloseable