Class FileColumnarStateStore

java.lang.Object
io.krabka.streams.columnar.FileColumnarStateStore
All Implemented Interfaces:
ColumnarStateStore
public final class FileColumnarStateStore extends Object implements ColumnarStateStore
Stores each partition snapshot in one atomically replaced local file.

The running state of a partition lives in <directory>/partition-<n>.snapshot and a barrier cut's state lives in <directory>/partition-<n>-epoch-<e>.snapshot. Saving writes a temporary file and moves it over the target — atomically where the file system supports it — so a crash mid-save leaves the previous snapshot intact. The file format is versioned and validated; a corrupt or truncated file fails loading with ColumnarException rather than restoring partial state.

The container is the layout the barrier design freezes for all three krabka streams libraries: a big-endian u32 version of 1, a u32 entry count, and then per entry a u32 name length, the UTF-8 name, a u32 value length, and the value bytes. Entries are written in ascending byte order of the name, so the same snapshot always produces the same bytes.

Use this store when each group member has its own local disk and partitions return to the same member across restarts, or accept rebuilding state from source topics after a member change.

Example

var stateStore = new FileColumnarStateStore(Path.of("/var/lib/app/state"));
var runner = ColumnarRunner.group(
    topology, consumer, producer,
    ColumnarErrorPolicy.fail(), stateStore, new ColumnarMetrics());

  • Constructor Details

    • FileColumnarStateStore

      public FileColumnarStateStore(Path directory)
      Creates a store rooted at a directory.

      The directory is created on the first save; it does not need to exist yet.

      Parameters:
      directory - the directory the snapshot files live in

    • FileColumnarStateStore

      public FileColumnarStateStore(Path directory, int retainedEpochs)
      Creates a store that keeps the newest retainedEpochs barrier snapshots per partition.
      Parameters:
      directory - the directory the snapshot files live in
      retainedEpochs - the positive number of barrier epochs to retain

    • FileColumnarStateStore

      public FileColumnarStateStore(Path directory, Duration retainedFor)
      Creates a store that keeps barrier snapshots newer than retainedFor.
      Parameters:
      directory - the directory the snapshot files live in
      retainedFor - the positive snapshot retention duration

  • Method Details

    • load

      public Map<String,byte[]> load(int partition, long epoch)
      Loads a partition's snapshot file for an epoch.
      Specified by:
      load in interface ColumnarStateStore
      Parameters:
      partition - the logical partition number
      epoch - the barrier epoch, or ColumnarStateStore.LIVE_EPOCH for the running state
      Returns:
      operator name to snapshot bytes; empty when no file exists
      Throws:
      ColumnarException - if the file exists but cannot be read or is corrupt

    • save

      public void save(int partition, long epoch, Map<String,byte[]> snapshot)
      Saves a partition's snapshots for an epoch, replacing the previous file atomically.
      Specified by:
      save in interface ColumnarStateStore
      Parameters:
      partition - the logical partition number
      epoch - the barrier epoch, or ColumnarStateStore.LIVE_EPOCH for the running state
      snapshot - operator name to snapshot bytes
      Throws:
      ColumnarException - if the file cannot be written or moved into place

    • retainedEpochs

      public Optional<List<Long>> retainedEpochs(int partition)
      Description copied from interface: ColumnarStateStore
      Lists the barrier epochs currently available for a partition.

      An empty optional means the store cannot enumerate epochs.

      Specified by:
      retainedEpochs in interface ColumnarStateStore
      Parameters:
      partition - the logical partition number
      Returns:
      the retained epochs in ascending order, when enumeration is supported

    • retain

      public ColumnarStateStore.EpochLease retain(long epoch)
      Description copied from interface: ColumnarStateStore
      Prevents an epoch from being reclaimed while a restore is reading it.
      Specified by:
      retain in interface ColumnarStateStore
      Parameters:
      epoch - the barrier epoch in use
      Returns:
      a lease that releases the epoch when closed