Producers

From a one-line blocking produce to idempotent batching and exactly-once transactions, all over gRPC.

A producer appending records while a consumer receives them from the same three-broker cluster.

The Java client lives in the jbroker.broker.client package of broker-core. See Install & deploy for adding the artifact to your build and starting a cluster; the examples below assume brokers at localhost:9092–9094.

Choosing a client

Two entry points, one wire protocol:

Both clients run a protocol-version handshake (ApiVersions) on first use of each connection, so a version mismatch fails up front rather than mid-request (see Error handling).

ClusterClient retry tuning is a ClusterClient.Config:

Minimal produce

The simplest path is a blocking single-record produce on BrokerClient. produce returns after the leader appends (acks=1); produceAcksAll blocks until every in-sync replica has the record (acks=all) and throws if replication cannot complete — the durable choice.

import java.nio.charset.StandardCharsets;
import jbroker.broker.client.BrokerClient;

try (var client = new BrokerClient("localhost", 9092)) {
    client.createTopic("orders", 3, 3);

    // Leader-only ack: fast, but a leader crash right after the ack can lose it.
    long offset = client.produce("orders", 0, "created".getBytes(StandardCharsets.UTF_8));

    // acks=all: returns only once the whole ISR has the record.
    long durable = client.produceAcksAll("orders", 0, "paid".getBytes(StandardCharsets.UTF_8));
}

produceBatch / produceBatchAcksAll take a List<byte[]> and ship all values in one RPC — one record per RPC is round-trip-bound, so batching is the first throughput lever. For applications, prefer BatchingProducer below, which does the batching for you and adds idempotence.

BatchingProducer

BatchingProducer is the asynchronous producer: send enqueues a record and returns a future immediately; a background sender thread packs records into per-partition batches and ships each batch with one acks=all RPC. A batch closes when its encoded size reaches batchSizeBytes or when lingerMs has passed since its first record, whichever comes first.

import java.util.List;
import java.util.concurrent.CompletableFuture;
import jbroker.broker.client.BatchingProducer;
import jbroker.broker.client.ClusterClient;

try (var cluster = new ClusterClient(List.of("localhost:9092", "localhost:9093", "localhost:9094"));
        var producer = BatchingProducer.create(cluster)) {
    CompletableFuture<Long> first = producer.send("orders", 0, "a".getBytes());
    CompletableFuture<Long> second = producer.send("orders", 0, "b".getBytes());
    producer.flush(); // force pending batches out and wait
    System.out.println("appended at " + first.join() + " and " + second.join());
}

Config knobs

The futures contract

Delivery is idempotent: the producer allocates a producer id lazily on first use and stamps every batch with a per-partition base sequence. A failed RPC is retried with the same sequence, which the broker deduplicates. So a future that completes normally means the record is on the partition exactly once, at the reported offset, replicated to the full ISR (every batch ships acks=all). Batches for one partition are sent strictly in order and never pipelined, which is what keeps the broker's contiguous-sequence check — and therefore the dedup guarantee — intact.

With the ClusterClient-backed producer this contract holds through leader failover: the same base sequence is retried against the new leader, which either dedupes the batch (it replicated before the old leader died) or appends it fresh.

zstd compression

Set Compression.ZSTD in the config to compress each batch's records section client-side. The broker keeps the codec when it re-encodes the batch for its log — compressed on the wire, compressed on disk, decompressed transparently on fetch. Compression is applied on the BrokerClient-backed sender; the cluster-routed sender currently ships batches uncompressed.

import jbroker.storage.Compression;

var config = new BatchingProducer.Config(
        64 * 1024, /*lingerMs*/ 5, /*deliveryTimeoutMs*/ 120_000, /*retryBackoffMs*/ 100,
        Compression.ZSTD);
try (var client = new BrokerClient("localhost", 9092);
        var producer = BatchingProducer.create(client, config)) {
    producer.send("orders", 0, largeJsonPayload);
    producer.flush();
}

TransactionalProducer

TransactionalProducer adds atomic multi-partition produces plus consumer-offset commits on top of ClusterClient — the client half of consume-transform-produce exactly-once. Records produced inside a transaction become visible to read_committed consumers only when the transaction commits; an abort makes them permanently invisible.

Lifecycle

import jbroker.broker.client.TransactionalProducer;

try (var producer = new TransactionalProducer(cluster, "checkout-svc-1")) {
    producer.initTransactions();       // adopt (producerId, epoch); fences older holders of this id

    producer.beginTransaction();
    producer.send("orders", 0, orderBytes);   // registers the partition on first touch
    producer.send("billing", 1, chargeBytes); // both commit or neither does
    producer.commitTransaction();
}

The transactionalId ("checkout-svc-1") is the stable identity: a restarted instance calling initTransactions() with the same id bumps the epoch, which fences any zombie still holding the old epoch and aborts whatever it left in flight. commitTransaction() returns once the coordinator has durably logged the decision; marker delivery continues broker-side and decides visibility.

Consume-transform-produce

sendOffsetsToTransaction stages the consumer group's offsets inside the transaction, so "output written" and "input marked consumed" commit or vanish together — the exactly-once loop:

import java.util.HashMap;
import jbroker.proto.common.TopicPartition;

var records = consumer.poll(Duration.ofMillis(500));
if (!records.isEmpty()) {
    producer.beginTransaction();
    var offsets = new HashMap<TopicPartition, Long>();
    for (var rec : records) {
        producer.send("orders-enriched", rec.tp().getPartition(), transform(rec.value()));
        offsets.merge(rec.tp(), rec.offset() + 1, Math::max); // commit points past each record
    }
    producer.sendOffsetsToTransaction("pipeline", offsets);
    producer.commitTransaction();
}

The staged offsets become visible to the group's FetchOffsets only on commit. If the transaction aborts, the group's committed position is untouched and the input is re-read — and since the aborted output is invisible to read_committed readers, reprocessing produces no duplicates downstream.

The transact() retry loop

transact(Runnable) wraps a whole attempt in the contract's abort-and-retry loop: begin, run the body, commit. Any abortable failure inside the body aborts the attempt, re-inits (the epoch bump both fences the aborted attempt's broker-side residue and resets sequences, so a resend can never be deduped against aborted data), and re-runs the body after a backoff:

producer.initTransactions();
producer.transact(() -> {
    producer.send("orders", 0, payload);
    producer.send("audit", 0, auditEntry);
});

Because aborted attempts are invisible to read_committed consumers, the retry is exactly-once end to end. A failure of the commit itself is never retried into an abort — the decision may already be logged coordinator-side and a decision is never reversed — it propagates to the caller.

Error taxonomy

Config knobs

Error handling

Every broker reply carries an error envelope with a numeric code. ClusterClient partitions the codes into two classes:

• Retriable — conditions that heal on their own: NOT_LEADER, NOT_ENOUGH_REPLICAS, STORAGE_FULL, QUOTA_VIOLATED, NOT_COORDINATOR, COORDINATOR_NOT_AVAILABLE. These are retried with backoff until the call deadline; exhaustion raises ClusterClient.RetriesExhaustedException with the last failure as its cause.
• Fail-fast — everything else (UNKNOWN_TOPIC, MESSAGE_TOO_LARGE, UNAUTHORIZED, ...): retrying the same request can never succeed, so the call throws ClusterClient.BrokerErrorException immediately. Its code() accessor exposes the numeric code for branching without parsing messages.

import jbroker.broker.ErrorCodes;
import jbroker.broker.client.UnsupportedBrokerException;

try {
    cluster.createTopic("orders", 3, 3);
} catch (ClusterClient.BrokerErrorException e) {
    if (e.code() != ErrorCodes.TOPIC_ALREADY_EXISTS) throw e; // already there: fine
} catch (ClusterClient.RetriesExhaustedException e) {
    // deadline spent; e.getCause() is the last failure seen
    throw e;
} catch (UnsupportedBrokerException e) {
    // protocol range mismatch: broker [min,max] vs client [min,max]
    System.err.println("upgrade needed for " + e.endpoint());
    throw e;
}

Version mismatch. The first use of each broker connection runs the ApiVersions handshake. A broker whose advertised protocol range does not overlap the client's raises UnsupportedBrokerException before any real RPC — fatal and never retried, with endpoint(), brokerMin()/brokerMax(), and clientMin()/clientMax() accessors for diagnostics. A failed handshake caches nothing, so a broker restarted with a compatible version needs no client restart.

Quota denials. A produce that exceeds a configured quota is rejected with QUOTA_VIOLATED; the message states the quota and how long to wait (for example produce quota exceeded: 1048576 B/s; retry in 250ms). ClusterClient treats it as retriable and backs off automatically; single-endpoint BrokerClient callers see the error and should wait the indicated interval before resending.

Python producer

The Python reference client (clients/python) speaks the same protos and record-batch bytes from a second language. It is deliberately minimal — synchronous, single-endpoint, uncompressed — with one convenience: a NOT_LEADER answer carrying suggested-leader hints gets one bounded retry against the hinted broker; everything else raises BrokerError with the numeric code and hint map attached. See the package README for install and stub generation.

from jbroker import Producer

with Producer("127.0.0.1", 9092) as producer:
    # One value or a list of values sent as a single batch.
    base, last = producer.produce("orders", 0, [b"one", b"two", b"three"])
    print(f"appended offsets {base}..{last}")

    # acks=-1 blocks until every ISR member has the batch.
    base, last = producer.produce("orders", 0, b"durable", acks=-1)

Next

Reading these records back — groups, commits, read_committed, dead-letter routing — is covered in Consumers.