API reference

Every public type in 1.4.2, grouped by module. Types not listed here are package-private implementation details and are not part of the compatibility surface.

The full Javadoc is browsable at https://krabka-io.github.io/krabka-streams-java/. Javadoc is also published alongside each artifact (-javadoc.jar) and is generated with Xdoclint:all plus -Werror, so every public type and member carries a complete comment. Every public type and package summary includes a runnable-style usage example.

`krabka-streams`

Package io.krabka.streams.

KrabkaStreamsConfig

public final class. A utility class that cannot be instantiated.

Member	Signature
`GROUP_PROTOCOL_CONFIG`	`public static final String` = `"group.protocol"`
`STREAMS_GROUP_PROTOCOL`	`public static final String` = `"streams"`
`withDefaults`	`public static Properties withDefaults(Map<?, ?> settings)`

withDefaults copies settings into a new Properties and applies krabka defaults with putIfAbsent. Throws NullPointerException for a null argument.

Everything else in this module is the Apache Kafka Streams 4.3.1 API, re-exported as an api dependency.

`krabka-streams-schema-serde`

Package io.krabka.streams.schema.

KrabkaSchemaRegistryClient

public final class. An asynchronous Confluent Schema Registry REST client.

Member	Signature
constructor	`KrabkaSchemaRegistryClient(URI baseUri)`
constructor	`KrabkaSchemaRegistryClient(URI baseUri, HttpClient httpClient, ObjectMapper objectMapper)`
constructor	`KrabkaSchemaRegistryClient(URI baseUri, String username, String password)`
constructor	injected client plus `int maxRetries`
`register`	`CompletableFuture<Integer> register(String subject, SchemaKind kind, String schema, String messageType)`
`lookup`	`CompletableFuture<Integer> lookup(String subject, SchemaKind kind, String schema, String messageType)`
`latest`	`CompletableFuture<RegisteredSchema> latest(String subject)`
`latestId`	`CompletableFuture<Integer> latestId(String subject)`
`schemaById`	`CompletableFuture<FetchedSchema> schemaById(int schemaId)`
management	subjects, versions, compatibility, deletion, and reference resolution methods

Nested records:

public record SchemaReference(String name, String subject, int version) {}

public record RegisteredSchema(..., List<SchemaReference> references) {}

public record FetchedSchema(String schema, String messageType, List<SchemaReference> references) {}

Failures complete the future exceptionally with SchemaRegistryException.

SchemaCache

public final class. A thread-safe store of resolved schema IDs and writer schemas.

Member	Signature
constructor	`SchemaCache(KrabkaSchemaRegistryClient client)`, which uses `AUTO_REGISTER` and `TopicNameStrategy`
constructor	`SchemaCache(KrabkaSchemaRegistryClient client, RegisterMode registerMode, SubjectNameStrategy subjectNameStrategy)`
`subject`	`String subject(String topic, Role role)`
`intern`	`void intern(String subject, SchemaKind kind, String schema, String messageType)`, idempotent by subject
`prewarm`	`CompletableFuture<Void> prewarm()`
`prewarmReport`	`CompletableFuture<PrewarmReport> prewarmReport()`
`idForSubject`	`OptionalInt idForSubject(String subject)`
`writerSchema`	`String writerSchema(int schemaId)`, which throws `SchemaFetchPendingException` on a miss
`writerMessageType`	`String writerMessageType(int schemaId)`, `null` when unknown
`writerReferences`	`Map<String, String> writerReferences(int schemaId)`
`seedSubjectId`	`void seedSubjectId(String subject, int schemaId)`
`seedWriterSchema`	`void seedWriterSchema(int schemaId, String schema)`
`seedWriterMessageType`	`void seedWriterMessageType(int schemaId, String messageType)`

AvroSerde<T>

public final class ... implements Serde<T>

Member	Signature
`forValue`	`static <T extends SpecificRecord> AvroSerde<T> forValue(Class<T> type, SchemaCache cache)`
`forKey`	`static <T extends SpecificRecord> AvroSerde<T> forKey(Class<T> type, SchemaCache cache)`
`generic`	`static AvroSerde<GenericRecord> generic(Schema schema, SchemaCache cache, Role role)`
`reflect`	`static <T> AvroSerde<T> reflect(Class<T> type, SchemaCache cache, Role role)`
`registerSubject`	`void registerSubject(String topic)`
`serializer` / `deserializer`	from `Serde<T>`

Registers the Avro canonical parsing form. Deserialization performs writer/reader schema resolution.

ProtobufSerde<T extends Message>

public final class ... implements Serde<T>

Member	Signature
`forValue`	`static <T extends Message> ProtobufSerde<T> forValue(T defaultInstance, SchemaCache cache)`
`forKey`	`static <T extends Message> ProtobufSerde<T> forKey(T defaultInstance, SchemaCache cache)`
`registerSubject`	`void registerSubject(String topic)`

Uses the Protobuf message-index framing and verifies the writer’s messageType.

JsonSchemaSerde<T>

public final class ... implements Serde<T>

Member	Signature
`forValue`	`static <T> JsonSchemaSerde<T> forValue(Class<T> type, String schema, SchemaCache cache, boolean validate)`
`forKey`	`static <T> JsonSchemaSerde<T> forKey(Class<T> type, String schema, SchemaCache cache, boolean validate)`
`forValue`	`static <T> JsonSchemaSerde<T> forValue(Class<T> type, String schema, SchemaCache cache, boolean validate, ObjectMapper objectMapper)`
`registerSubject`	`void registerSubject(String topic)`

validate applies in both directions. $schema selects Draft 4, 6, 7, 2019-09, or 2020-12; an extended factory accepts an explicit dialect, strategy, and mapper.

ConfluentWireFormat

public final class. Framing helpers.

Member	Signature
`MAGIC`	`public static final byte` = `0`
`encode`	`static byte[] encode(int schemaId, byte[] body)`
`decode`	`static Frame decode(byte[] bytes)`
`encodeProtobuf`	`static byte[] encodeProtobuf(int schemaId, List<Integer> messageIndexes, byte[] body)`
`decodeProtobuf`	`static ProtobufFrame decodeProtobuf(byte[] bytes)`

public record Frame(int schemaId, byte[] body) {}

public record ProtobufFrame(int schemaId, List<Integer> messageIndexes, byte[] body) {}

Both records copy body on construction and on access.

LocalSchemaCompatibility

Network-free pairwise checks for BACKWARD, FORWARD, and FULL modes. avro and json accept schema text; protobuf accepts two FileDescriptor values. Every check returns Result(boolean compatible, List<String> incompatibilities).

Enums and interfaces

Type	Values / members
`Role`	`KEY`, `VALUE`
`SchemaKind`	`AVRO` (no wire name), `PROTOBUF`, `JSON`
`RegisterMode`	`AUTO_REGISTER`, `LOOKUP_ONLY`, `USE_LATEST`
`SubjectNameStrategy`	`@FunctionalInterface String subject(String topic, Role role)`
`TopicNameStrategy`	`implements SubjectNameStrategy`, producing `topic + "-key"` or `topic + "-value"`

Exceptions

Type	Extends	Members
`SchemaRegistryException`	`RuntimeException`	`SchemaRegistryException(String)`, `(String, Throwable)`, `(int statusCode, String body)`; `int statusCode()`, which returns `-1` for transport or parsing errors
`SchemaFetchPendingException`	`org.apache.kafka.common.errors.RetriableException`	`SchemaFetchPendingException(int schemaId)`; `int schemaId()`

`krabka-streams-columnar`

Package io.krabka.streams.columnar.

ColumnarTopology

public final class. The topology builder.

Member	Signature
constructor	`ColumnarTopology(BufferAllocator allocator)`
`addSource`	`ColumnarNode addSource(String name, Collection<String> topics, BatchCodec codec)`
`addOperator`	`ColumnarNode addOperator(String name, BuiltinOp operator, ColumnarNode parent)`
`addOperator`	`ColumnarNode addOperator(String name, Supplier<? extends ColumnarProcessor> processor, ColumnarNode parent)`
`addMerge`	`ColumnarNode addMerge(String name, Collection<ColumnarNode> parents)`
`addJoin`	`ColumnarNode addJoin(String name, ColumnarJoin join, ColumnarNode left, ColumnarNode right)`
`addSink`	`ColumnarNode addSink(String name, String topic, BatchCodec codec, ColumnarNode parent)`
`addPassThroughSink`	`ColumnarNode addPassThroughSink(String name, String topic, ColumnarNode source)`
`sourceTopics`	`List<String> sourceTopics()`
`validate`	`void validate()`, which throws `ColumnarException`
`build`	`BuiltColumnarTopology build()`

BuiltColumnarTopology

public final class. Validated, stateful, reusable, and synchronized.

Member	Signature
`runBatch`	`List<ProducedToTopic> runBatch(String topic, List<ConsumedRecord> records)`
`runBatches`	`List<ProducedToTopic> runBatches(Map<String, List<ConsumedRecord>> input)`
`runPartitionBatches`	evaluates one co-partitioned input map
state lifecycle	`snapshotPartition`, `restorePartition`, `releasePartition`, and `close`

ColumnarNode

public final class. An opaque parent handle with no public members.

ColumnarRunner

public final class. A utility class.

public static long runPartitionOnce(
    ColumnarTopology topology,
    Consumer<byte[], byte[]> consumer,
    Producer<byte[], byte[]> producer,
    String topic,
    int partition,
    long offset,
    Duration pollTimeout)

Commits and returns the next offset. group(...) creates a subscribed reusable runner; group runners provide ordinary and transactional runOnce methods. The full overload accepts ColumnarErrorPolicy, ColumnarStateStore, and ColumnarMetrics. sendAsync returns a CompletableFuture<Void> for broker acknowledgements.

Runner support types:

ColumnarErrorPolicy: fail(), skip(), or deadLetter(topic).
ColumnarMetrics: lock-free counters exposed through an immutable Snapshot.
ColumnarStateStore: load(int partition, long epoch) and save(int partition, long epoch, Map<String, byte[]> snapshot); LIVE_EPOCH is the rebalance state and none() is ephemeral.
FileColumnarStateStore: atomically replaced snapshot files under a caller-owned path, in the container the three krabka streams libraries share.

The seven-argument group overload also accepts a BarrierAlignment. A runner built with one exposes pendingCut(), restoreToEpoch(long), and restoreToLatestCut().

Codecs

Type	Signature
`BatchCodec`	`interface`: decode/encode methods, with default topic-aware overloads
`BlobCodec`	`final class implements BatchCodec`: `BlobCodec(BufferAllocator)`, `BlobCodec(BufferAllocator, int maxRecordBytes)`
`GzipBatchCodec`	`final class implements BatchCodec`: bounded GZIP decorator for another codec
`RowCodec<T>`	`final class implements BatchCodec`: `RowCodec(Serde<T> valueSerde, RowBridge<T> rowBridge, BufferAllocator allocator)`
`ArrowIpcSerde`	`final class implements Serde<VectorSchemaRoot>`: `ArrowIpcSerde(BufferAllocator)`

BlobCodec constants: DEFAULT_MAX_RECORD_BYTES (900 * 1024), KEY_COLUMN (__key), TIMESTAMP_COLUMN (__timestamp), PARTITION_COLUMN (__partition), OFFSET_COLUMN (__offset), HEADERS_COLUMN (__headers), plus payloadColumn(String) for escaped collisions.

Row bridges

Type	Signature
`RowBridge<T>`	`interface`: `VectorSchemaRoot rowsToBatch(List<T> rows, BufferAllocator)`, `List<T> batchToRows(VectorSchemaRoot)`
`JsonRowBridge<T>`	constructors accept `Class<T>`, optional `ObjectMapper` or Arrow `Schema`; `fromJsonSchema` derives fields

ArrowValues

public final class of static helpers for RowBridge and ColumnarProcessor implementations, exposing the same type-coercing conversions the built-in operators use:

static VectorSchemaRoot createRoot(List<Field> fields, int rows, BufferAllocator allocator)
static void set(FieldVector vector, int row, Object value)
static Object get(FieldVector vector, int row)
static void finish(VectorSchemaRoot root)

Operators

Type	Signature
`ColumnarProcessor`	`@FunctionalInterface void process(ColumnarContext context, VectorSchemaRoot batch)`
`StatefulColumnarProcessor`	processor plus `snapshot()` and `restore(byte[])`
`ColumnarContext`	`final class`: `void forward(VectorSchemaRoot batch)`
`RowPredicate`	`@FunctionalInterface boolean test(VectorSchemaRoot batch, int row)`
`RowValue`	`@FunctionalInterface Object value(VectorSchemaRoot batch, int row)`
`DerivedColumn`	`record DerivedColumn(Field field, RowValue value)`
`Aggregation`	`record Aggregation(String inputColumn, String outputColumn, AggregateFunction function, ArrowType outputType)`
`AggregateFunction`	`enum`: `COUNT`, `SUM`, `MIN`, `MAX`

BuiltinOp is a public final class implements ColumnarProcessor:

static BuiltinOp filter(BufferAllocator allocator, RowPredicate predicate)
static BuiltinOp select(BufferAllocator allocator, String... columns)
static BuiltinOp withColumns(BufferAllocator allocator, DerivedColumn... columns)
static BuiltinOp groupBy(
    BufferAllocator allocator, Collection<String> keys, Aggregation... aggregations)
static BuiltinOp windowedGroupBy(
    BufferAllocator allocator,
    Collection<String> keys,
    Duration size,
    Aggregation... aggregations)
static BuiltinOp windowedGroupBy(
    BufferAllocator allocator,
    Collection<String> keys,
    Duration size,
    Duration retention,
    Aggregation... aggregations)
void process(ColumnarContext context, VectorSchemaRoot batch)

Records and exceptions

public record RecordHeader(String key, byte[] value) {}

public record ConsumedRecord(..., List<RecordHeader> headers) {}

public record ProduceRecord(..., List<RecordHeader> headers) {}

public record ProducedToTopic(String topic, ProduceRecord record) {}

public record ColumnarJoin(
    String leftKey, String rightKey, Duration window, String leftPrefix, String rightPrefix) {}

public final class ColumnarException extends RuntimeException {
  public ColumnarException(String message);

  public ColumnarException(String message, Throwable cause);
}

ConsumedRecord and ProduceRecord copy their byte arrays on construction and on access. value must not be null; key may be.

Barrier alignment

Package io.krabka.streams.columnar.barrier. See Barrier alignment for usage.

Type	Signature
`BarrierCut`	`record BarrierCut(String group, long epoch, long triggeredAt, long completedAt, BarrierCutStatus status, Map<TopicPartition, Long> offsets, Set<TopicPartition> missing)`
`BarrierCutStatus`	`enum`: `COMPLETE`, `PARTIAL`; `code()` and `fromCode(int)`
`BarrierCutDecoder`	`final class`: `static Optional<BarrierCut> decode(byte[] key, byte[] value)`; `TOPIC` is `__barrier_state`
`BarrierCutReader`	`final class`: `BarrierCutReader(Consumer<byte[], byte[]>)`, optional `Duration`; `latestCompleteCut(String)` and `completeCutsAfter(String, long)`
`BarrierAlignment`	`record BarrierAlignment(String group, BarrierCutReader reader, BarrierListener listener)`; `on(String, BarrierCutReader)` and `withListener`
`BarrierListener`	`interface`: `void onBarrier(BarrierCut cut)`; `none()`

BarrierCut members: complete(), offset(TopicPartition) returning OptionalLong, recordsBefore(TopicPartition, List<ConsumedRecord>), and reached(TopicPartition, long position). The offsets and the missing partitions are copied on construction, so a decoded cut is immutable.

`krabka-streams-columnar-schema`

Package io.krabka.streams.columnar.schema. See Columnar processing for usage. The module depends on both krabka-streams-columnar and krabka-streams-schema-serde, plus protobuf-java-util as an implementation dependency for the canonical Protobuf JSON fallback.

AvroBatchCodec

public final class implements BatchCodec

Member	Signature
`generic`	`static AvroBatchCodec generic(Schema schema, SchemaCache cache, BufferAllocator allocator)`
`forValue`	`static <T extends SpecificRecord> AvroBatchCodec forValue(Class<T> type, SchemaCache cache, BufferAllocator allocator)`
`registerSubject`	`void registerSubject(String topic)`
`arrowSchema`	`Schema arrowSchema()` (Arrow), the payload columns without metadata columns
decode/encode	the four `BatchCodec` methods

ProtobufBatchCodec

public final class implements BatchCodec

Member	Signature
`of`	`static <T extends Message> ProtobufBatchCodec of(T defaultInstance, SchemaCache cache, BufferAllocator allocator)`
`registerSubject`	`void registerSubject(String topic)`
`arrowSchema`	`Schema arrowSchema()` (Arrow), the payload columns without metadata columns
decode/encode	the four `BatchCodec` methods

Row bridges and schema converters

Type	Signature
`AvroRowBridge<T extends IndexedRecord>`	`implements RowBridge<T>`: `generic(Schema)`, `forSpecific(Class<T>)`, `arrowSchema()`
`ProtobufRowBridge<T extends Message>`	`implements RowBridge<T>`: `of(T defaultInstance)`, `arrowSchema()`
`AvroArrowSchemas`	`static Schema toArrowSchema(org.apache.avro.Schema)`, `static Field toArrowField(String, org.apache.avro.Schema)`
`ProtobufArrowSchemas`	`static Schema toArrowSchema(Descriptor)`, `static Field toArrowField(FieldDescriptor)`

The bridges derive their Arrow schema once, at construction, from the reader schema or message descriptor; mid-stream writer evolution never changes the columns. Conversion choices that the Arrow type alone cannot reverse are tagged with krabka.avro.*, krabka.proto.*, or krabka.json field metadata.

`krabka-streams-coordination`

Package io.krabka.streams.coordination. See Coordination for usage. The module depends on krabka-streams for the pinned Kafka client version and uses the producer, consumer, and admin clients only.

Role, MemberId

public final class, each Comparable and each built by one factory.

Member	Signature
`MAX_LENGTH`	`public static final int` = `249` bytes, on both types
`of`	`static Role of(String name)`, `static MemberId of(String id)`
accessor	`String name()` on `Role`, `String id()` on `MemberId`
`bytes`	`byte[] bytes()` on `Role`, the UTF-8 name the partition rule hashes

Both factories throw CoordinationException for an empty name and for a name past the byte bound. A Role becomes a Kafka transactional.id.

FencingToken

public final class implements Comparable<FencingToken>

Member	Signature
`NO_EPOCH`	`public static final FencingToken`, the role no member has ever taken
`of`	`static FencingToken of(long producerId, short producerEpoch)`
`parse`	`static FencingToken parse(String text)`, the `id:epoch` form
`producerId`	`long producerId()`
`producerEpoch`	`short producerEpoch()`
`minted`	`boolean minted()`
`compareTo`	lexicographic: producer id first, then producer epoch
`supersedes`	`boolean supersedes(FencingToken other)`

of rejects a negative value in either position.

Record types and codec

Type	Signature
`RecordKind`	`enum`: `REGISTRATION` (0), `LEASE` (1); `code()`, `fromCode(short)`
`CoordinationValue`	`sealed interface permits Registration, Lease`: `kind()`, `member()`
`Registration`	`record(MemberId member, long registeredAt)`
`Lease`	`record(MemberId member, FencingToken token, long grantedAt, long deadline)`
`CoordinationKey`	`record(RecordKind kind, Role role, Optional<MemberId> member)`
`CoordinationEntry`	`record(long offset, CoordinationKey key, Optional<CoordinationValue> value)`
`CoordinationCodec`	`TOPIC`, `encodeKey`, `encodeValue`, `decodeKey`, `decodeValue`, `isTombstone`
`RolePartitioner`	`DEFAULT_PARTITIONS` = 16; `static int partitionFor(Role role, int partitions)`

CoordinationKey.registration(Role, MemberId) and CoordinationKey.lease(Role) build a key that keeps the member rule of the frozen layout. decodeValue returns an empty Optional for a tombstone.

Lease policy and clock

Type	Signature
`LeaseConfig`	`defaults()`, `of(Duration, Duration, Duration)`, `duration()`, `renewInterval()`, `challengeStagger()`, `renewsWithMargin()`
`LeaseConfig`	`challengeDelayMillis(int rank)`, `grant(MemberId, FencingToken, long)`, `timing(Lease)`
`LeaseTiming`	`lease()`, `expiresAtMillis()`, `liveAt(long)`, `remainingAt(long)`, `renewAtMillis()`, `renewDueAt(long)`
`LeaseTiming`	`challengeAtMillis(int rank)`
`Clock`	`long nowMillis()`; `static Clock system()`
`ManualClock`	`ManualClock(long nowMillis)`, `set(long)`, `advance(Duration)`

The defaults are a 30-second lease, a 10-second renew interval, and a 5-second challenge stagger. of rejects an extent that is not positive and a renew interval that is not shorter than the duration.

Succession

Type	Signature
`RosterEntry`	`record(MemberId member, long offset, long registeredAt)`
`RoleState`	`empty()`, `fromRecords(Role, Iterable<CoordinationEntry>)`
`RoleState`	`roster()`, `lease()`, `holder()`, `entry(MemberId)`, `rankOf(MemberId)`
`RoleStateBuilder`	`RoleStateBuilder(Role)`, `role()`, `apply(CoordinationEntry)`, `build()`
`Decision`	`record(Decision.Action action, long waitUntilMillis)`
`Decision.Action`	`NOT_REGISTERED`, `HOLD`, `CHALLENGE`, `WAIT`
`Succession`	`static Decision evaluate(RoleState, MemberId, long nowMillis, LeaseConfig)`

The rules perform no input and no output. rankOf returns an OptionalInt, and the rank is the index in the roster after the removal of the current holder.

Transport and client

Type	Signature
`CoordinationTransport`	`acquireEpoch`, `readRoleRecords`, `register`, `writeLease`, `clearLease`, `describe`, `close`
`KafkaCoordinationTransport`	`(Admin, Producer, Function<Role, Producer>, Consumer)` and an overload that takes a poll timeout
`CoordinationClient`	`(CoordinationTransport)` and `(CoordinationTransport, LeaseConfig, Clock, Duration)`
`CoordinationClient`	`config()`, `readState(Role)`, `describe(Role)`, `tryAcquire(Role, MemberId)`, `acquire(...)`
`Leadership`	`role()`, `member()`, `token()`, `lease()`, `timing()`, `held()`, `renewDue()`, `renew()`, `resign()`
`LeadershipStatus`	`record(Role role, FencingToken token, RoleState state)`; `holder()`, `lease()`, `held()`, `current()`
`CoordinationException`	`public class extends RuntimeException`
`FencedException`	`public final class extends CoordinationException`; `role()`

Leadership is AutoCloseable and resigns on close. FencedException is the only signal that a leadership ended.

`krabka-streams-test-utils`

Package io.krabka.streams.test. See Testing for usage.

ColumnarTestDriver

public final class

Member	Signature
constructor	`ColumnarTestDriver(BuiltColumnarTopology topology)`
`pipeInput`	overloads accept record bytes and optional `List<RecordHeader>`
`pipeBatch`	`void pipeBatch(String topic, List<ConsumedRecord> records)`
`failNext`	injects one `RuntimeException` before the next batch evaluation
`isOutputEmpty`	`boolean isOutputEmpty(String topic)`
`outputSize`	`int outputSize(String topic)`
`readOutput`	`ProduceRecord readOutput(String topic)`, which throws `NoSuchElementException`
`drainOutput`	`List<ProduceRecord> drainOutput(String topic)`

SchemaRegistryStub

public final class ... implements AutoCloseable

Member	Signature
constructor	`SchemaRegistryStub() throws IOException`, which binds `127.0.0.1` on an ephemeral port
`uri`	`URI uri()`
`requestCount`	`int requestCount(String method, String path)`
`close`	`void close()`

This module also re-exports Apache Kafka’s kafka-streams-test-utils, including TopologyTestDriver.

Compatibility

Nothing is deprecated yet. The public surface above is what future releases will be judged against; package-private types (AbstractSchemaSerde, ArrowBatchSupport, BridgeMetadata, ProtobufSchemaPrinter) may change at any time.

Transitive api dependencies are part of the surface too: Kafka Streams 4.3.1, Avro 1.12.1, Protobuf 4.33.5, Jackson 2.22.0, and Arrow 19.0.0.