krabka streams for Java

krabka-streams-java is the Java client library for stream processing with krabka. It uses the Apache Kafka Streams API and adds krabka schema registry and Apache Arrow support.

The minimum Java version is 17.

implementation("io.krabka:krabka-streams:1.4.0")

Modules

Artifact	Purpose
io.krabka:krabka-streams	Apache Kafka Streams API and krabka defaults
io.krabka:krabka-streams-schema-serde	Avro, Protobuf, and JSON Schema serdes
io.krabka:krabka-streams-columnar	Apache Arrow batch processing
io.krabka:krabka-streams-columnar-schema	Avro and Protobuf Arrow bridges
io.krabka:krabka-streams-coordination	Leader election, leases, fencing tokens
io.krabka:krabka-streams-test-utils	Test helpers for all modules
io.krabka:krabka-streams-bom	Version constraints for every module

Each module depends on krabka-streams, so any one of them puts the Kafka Streams API on your classpath at the version this release pins.

Documentation

Full documentation is in docs/. The API reference for the latest release is published at https://krabka-io.github.io/krabka-streams-java/.

Document	Contents
Getting started	Requirements, coordinates, and first examples
Configuration	KrabkaStreamsConfig, broker requirements, JVM flags
Schema registry	Registry client, schema cache, prewarming
Serdes	Avro, Protobuf, JSON Schema, and the Confluent wire format
Columnar processing	Arrow batches, codecs, topologies, runner
Columnar operators	Built-in operators and buffer ownership
Barrier alignment	Cuts, aligned processing, epoch-keyed snapshots, restore
Coordination	Leader election, leases, fencing tokens, succession
Testing	Test drivers, registry stub, integration suite
API reference	Every public type
Architecture	Module layout and design decisions
Runtime constraints	Broker, JVM, Arrow, and packaging constraints
Troubleshooting	Error messages mapped to causes
Build and release	Gradle tasks, CI, publishing

Kafka Streams

The Kafka Streams DSL, Processor API, state stores, and interactive queries are exported unchanged. The only krabka-specific step is the configuration helper, which enables the streams group protocol.

var settings =
    Map.<String, Object>of(
        StreamsConfig.APPLICATION_ID_CONFIG, "order-counter",
        StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

var streams = new KafkaStreams(topology, KrabkaStreamsConfig.withDefaults(settings));

Settings you provide are never overwritten. See Configuration.

Build

./gradlew build
bazel build //...
bazel test //...

On Windows, use gradlew.bat build.

To run Bazel builds on BuildBuddy RBE, create an ignored user.bazelrc containing your BuildBuddy API key:

build --remote_header=x-buildbuddy-api-key=YOUR_API_KEY

Then enable the checked-in remote configuration:

bazel test //... --config=remote

To consume the source directly from another Bazel module, add this to its MODULE.bazel (replace the commit with the revision you want to pin):

bazel_dep(name = "krabka_streams_java", version = "1.4.0")
git_override(
    module_name = "krabka_streams_java",
    remote = "https://github.com/krabka-io/krabka-streams-java.git",
    commit = "<commit SHA>",
)

Then depend on any public module target:

deps = ["@krabka_streams_java//krabka-streams:krabka-streams"]

Run the broker integration test against a ready broker:

KRABKA_INTEGRATION_BOOTSTRAP=localhost:9092 \
  ./gradlew :krabka-streams-test-utils:integrationTest

The broker must enable the streams group protocol and finalize streams.version=1. For Apache Kafka 4.3.1, set group.streams.num.standby.replicas=1 to run the standby check.

Schema registry example

var client = new KrabkaSchemaRegistryClient(URI.create("http://localhost:8081"));
var cache = new SchemaCache(client);
var serde = JsonSchemaSerde.forValue(Order.class, orderSchema, cache, true);

serde.registerSubject("orders");
cache.prewarm().join();

The cache resolves schema IDs before processing starts. If a consumer sees an unknown writer ID, the cache starts one background fetch and throws SchemaFetchPendingException. The exception is retriable.

See Schema registry and Serdes.

Arrow columnar processing

Columnar topologies use VectorSchemaRoot. Each fetched topic partition batch is one processing unit. The built-in operations are filter, select, with-columns, cumulative and windowed group-by, plus stateful event-time joins.

Arrow 19 needs this JVM option when it uses direct buffers:

--add-opens=java.base/java.nio=ALL-UNNAMED

The metadata columns are __key, __timestamp, __partition, __offset, and __headers. Colliding payload names are escaped and restored automatically. BlobCodec packs Arrow IPC output under a 900 KiB hard limit. Built topologies retain processor and aggregate state per logical partition across fetched batches. The group runner adds snapshots, rebalance hooks, metrics, acknowledged asynchronous sends, and skip or dead-letter error policies. GzipBatchCodec provides bounded per-record compression.

The group runner also aligns on the broker's barrier cuts. It reads each cut from the internal __barrier_state topic, holds every record at or above the cut back, snapshots each partition under the cut's epoch, and restores to that epoch on request.

krabka-streams-columnar-schema bridges the registry serdes into the columnar runtime: AvroBatchCodec and ProtobufBatchCodec decode registry-framed topics into batches whose columns follow the record schema — structs, lists, maps, decimals, and timestamps as native Arrow types — and encode processed batches back.

See Columnar processing, Columnar operators, and Barrier alignment.

Leader election

One role elects one leader, and Kafka's transaction coordinator supplies the proof. The leadership epoch is the producer epoch that the coordinator mints for transactional.id = <role>. The quorum mints it, the value only grows, and every broker rejects a write that carries an older one. The lease adds no safety. It decides when a standby stops waiting for a quiet holder, and nothing else.

Role role = Role.of("controller");
MemberId me = MemberId.of("node-1");
try (CoordinationClient client = new CoordinationClient(transport);
    Leadership leadership = client.acquire(role, me, Duration.ofMinutes(1))) {
  while (running) {
    if (leadership.renewDue()) {
      leadership.renew();
    }
    dispatch(leadership.token());
  }
} catch (FencedException lost) {
  controller.stop();
}

Per-role state lives in the compacted internal topic __coordination_state. A candidate appends a registration record, and the offset of that record is its place in the succession order. A recovered node registers again and lands at the tail, so it never preempts the member that replaced it.

See Coordination.

Test utilities

ColumnarTestDriver runs a built columnar topology without a broker. SchemaRegistryStub provides a stateful local implementation of the registry endpoints used by the serdes. The artifact also exports Apache Kafka's TopologyTestDriver for ordinary Kafka Streams topologies.

See Testing.

Status

The current version is 1.4.0. See PARITY.md for the parity checklist, CHANGELOG.md for release notes, and runtime constraints.

License

Apache License 2.0.