Kafka 4.2: Share Groups Unlock Per-Partition Parallelism, Streams Rebalancing Moves to Broker

Kafka Basics and the Partition Bottleneck

Kafka is a distributed commit log. Producers write messages to topics, which are split into partitions; messages are appended in arrival order. Brokers store and replicate these logs, while the controller handles leader election and metadata. Since version 4.0, metadata runs exclusively on KRaft, removing the need for a separate ZooKeeper ensemble.

Three core properties drive Kafka's adoption: high throughput, replayable logs via offsets, and consumer-managed progress. It sits between log collection, order events, metrics, and stream processing.

Traditional consumer groups enforce a strict rule: a single partition is assigned to only one consumer in the group at any moment. This guarantees per-partition order but serializes work within that partition. Tasks such as order processing, SMS sending, or calling a slow external API are bound by this single-threaded constraint — adding more consumers does not help if the partition count stays the same.

The 4.2 Release Line

Kafka 4.2.0 (released 2026-02-17) focused on new capabilities: Share Groups (KIP-932) graduated, and Kafka Streams server-side rebalancing (KIP-1071) became a core feature. 4.2.1 fixed a deadlock in the shared-group path and a severe issue in Streams offline migration. 4.2.2 (2026-09-29) is a maintenance release addressing roughly 33 issues, explicitly aimed at making the 4.2 capabilities production-ready. Therefore, "how strong is 4.2" must be evaluated across the whole line: new features in 4.2.0, production readiness at 4.2.2.

Minor but welcome improvements include unified --bootstrap-server across CLI tools, thread-safe concurrent reads of RecordHeader, Connect support for external JSON Schema (avoiding per-message schema overhead), and client compatibility with Java 25.

Share Groups: Multiple Consumers on One Partition

Share Groups, the most visible change in 4.2, implement KIP-932. Unlike consumer groups that assign whole partitions, Share Groups cooperate at the record level. Records from the same partition can be dispatched to multiple consumers simultaneously, each record protected by a time-limited fetch lock (default 30 seconds). The consumer must explicitly signal the outcome.

Four acknowledgment types cover every outcome: ACCEPT: processing succeeded. RELEASE: transient failure; return the record for redelivery to another consumer or a later retry. REJECT: the record is fundamentally invalid; never redeliver. RENEW (new in 4.2): processing continues beyond the lock timeout; extend the lock. This is designed for tasks that genuinely exceed 30 seconds.

The acknowledgment flow: lock → process → signal → coordinator. Share Groups suit independently completable work — notifications, tagging, external API calls — while strict per-partition ordering remains the domain of classic consumer groups. Both group types can coexist; choose per task without rewriting the entire pipeline.

Explicit Acknowledgment Mechanics

With explicit acknowledgment, acknowledge() records the decision locally; commitSync() or the next poll() flushes results to the cluster. The example below demonstrates an order-processing shared consumer: invalid orders are rejected, transient failures are released, successful ones accepted.

/**
 * Order shared consumer.
 * Orders in the same partition can be processed by multiple instances in the group.
 */
public class OrderShareConsumer {

    /** Pull orders and acknowledge each record by result. */
    public void consume() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("group.id", "order-workers");
        props.setProperty("key.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer",
                "org.apache.kafka.common.serialization.StringDeserializer");
        // Explicit acknowledgment: every record must receive ACCEPT, RELEASE, or REJECT
        props.setProperty("share.acknowledgement.mode", "explicit");
        // Strictly limit by max.poll.records to avoid grabbing too many slow orders
        props.setProperty("share.acquire.mode", "record_limit");
        props.setProperty("max.poll.records", "20");

        try (KafkaShareConsumer<String, String> consumer = new KafkaShareConsumer<>(props)) {
            consumer.subscribe(List.of("orders"));
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        handleOrder(record.value());
                        consumer.acknowledge(record, AcknowledgeType.ACCEPT);
                    } catch (IllegalArgumentException badData) {
                        consumer.acknowledge(record, AcknowledgeType.REJECT);
                    } catch (Exception temporary) {
                        consumer.acknowledge(record, AcknowledgeType.RELEASE);
                    }
                }
                consumer.commitSync();
            }
        }
    }

    /** Process one order. Throw IllegalArgumentException for invalid data. */
    private void handleOrder(String body) {
        if (body == null || body.isBlank()) {
            throw new IllegalArgumentException("empty order");
        }
        // Real business logic: deduct inventory, send SMS, write downstream
    }
}

Handling Long-Running Tasks with RENEW

External API calls often exceed 30 seconds. For such records, the consumer issues AcknowledgeType.RENEW inside the processing loop and continues to poll(). A successful renewal returns the same record in the next poll() result. Renewal blocks delivery progress for the whole group, so it should be reserved for genuinely slow records.

Kafka Streams: Closer to Production Ready

Kafka Streams performs stream processing inside Kafka: filtering, joins, windowing, writing results back to topics. Previously, rebalancing was computed client-side; any membership change stopped the entire group. 4.2 graduates KIP-1071: task assignment moves to the broker for continuous computation. Applications register as a streams group, gaining dedicated observability and troubleshooting entry points.

Enabling the new protocol requires a single configuration:

props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, "streams");

The 4.2.0 offline migration had a data-plane defect; 4.2.1 fixed it. Migrate to streams groups using 4.2.2 directly. New applications can simply adopt a fresh, unused application.id. Core capabilities are production-ready; however, major topology changes and pattern subscriptions should be verified against the official developer guide for support coverage.

Dead Letter Queues for Poison Pills (KIP-1034)

Previously, exceptions in processing often halted the entire stream or were silently swallowed in mapValues. KIP-1034 adds dead-letter support to the exception handler. Configure a DLQ topic; the built-in handler forwards the failed original record with headers containing the exception class, stack trace, source topic, and offset.

/**
 * Basic config for order stream.
 * Failed records go to order-dlq; main flow continues.
 */
public class OrderStreamConfig {

    /** Assemble Streams config with DLQ and new rebalancing protocol. */
    public static Properties create() {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-hourly");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.GROUP_PROTOCOL_CONFIG, "streams");
        props.put(StreamsConfig.ERRORS_DEAD_LETTER_QUEUE_TOPIC_NAME_CONFIG, "order-dlq");
        props.put(StreamsConfig.PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG,
                "org.apache.kafka.streams.errors.LogAndContinueProcessingExceptionHandler");
        return props;
    }
}

Custom handlers return Response.resume with a ProducerRecord to send to the DLQ, or Response.fail to stop the task.

Anchored Punctuation (KIP-1146)

Wall-clock scheduling previously made the first trigger dependent on process start time, making alignment to exact boundaries (e.g., hourly) difficult. Anchored punctuation lets you specify the first trigger instant; subsequent triggers follow at the defined interval.

/**
 * Settle hourly at exact hour boundaries.
 * startTime pins the first trigger moment.
 */
@Override
public void init(ProcessorContext<String, String> context) {
    context.schedule(
            Instant.parse("2026-10-26T00:00:00Z"),
            Duration.ofHours(1),
            PunctuationType.WALL_CLOCK_TIME,
            timestamp -> settleHourlyOrders());
}

Key Fixes in 4.2.2

The 4.2.2 release notes read like a repair log. Highlights that affect production judgment:

Exactly-once checkpoint cleanup : Checkpoints written at recovery end could persist into runtime. An unclean crash would skip state cleanup. This gap is closed.

Deadlock in KafkaStreams.removeStreamThread : Could collide with the "replace stream thread" exception handler, affecting applications that dynamically scale thread count.

Consumer group migration protocol issues : Groups with migration disabled could still accept classic-protocol members, getting stuck in a hybrid state. Also, large metadata from classic members could corrupt the coordinator state, preventing reload on restart. Both fixed.

ListDeserializer truncation handling : Previously emitted a partial record on truncated input; now treats it as an error.

Honest record-e2e-latency-max : Previously measured only fetch latency, omitting processing time, diverging from KIP-613. After upgrade, the metric includes processing time; alert thresholds must be revisited.

Re-authentication CPU spin : Consumers burned CPU after re-authentication failure.

Admin client timeout on missing leader : Admin client waited the full timeout when a cached partition leader had left the cluster.

Protocol decoding limits : Added allocation caps for oversized arrays and excessive header fields.

Dependency upgrades : jackson-databind, lz4-java, jline received security updates.

Running Share Groups Locally

Use the official image apache/kafka:4.2.2. On a single node, the share coordinator state topic replication factor must be set to 1, otherwise topic creation fails. The following Docker Compose (adapted from the official single-node example) starts a 4.2.2 broker with KRaft:

services:
  broker:
    image: apache/kafka:4.2.2
    hostname: broker
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: CONTROLLER://:29093,PLAINTEXT_HOST://:9092,PLAINTEXT://:19092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT_HOST://localhost:9092,PLAINTEXT://broker:19092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@broker:29093
      CLUSTER_ID: 4L6g3nShT-eMCtK--X86sw
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_SHARE_COORDINATOR_STATE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_SHARE_COORDINATOR_STATE_TOPIC_MIN_ISR: 1

Producers remain unchanged; 4.2 does not alter the producer API. The example OrderProducer sends orders to the orders topic.

/**
 * Write an order to the orders topic for shared consumers.
 */
public class OrderProducer {

    /** Send one order message. */
    public void send(String orderId, String body) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "localhost:9092");
        props.setProperty("key.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");
        props.setProperty("value.serializer",
                "org.apache.kafka.common.serialization.StringSerializer");

        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            producer.send(new ProducerRecord<>("orders", orderId, body));
            producer.flush();
        }
    }
}

Start two instances of the OrderShareConsumer with the same group.id. Both will pull different orders from the same partition concurrently.