Apache Kafka — Why Rebalance Storms Kill Consumer Groups

# ── Step 1: Create a topic with 3 partitions and replication factor 3 ──
# More partitions = more parallelism, but also more overhead.
# Rule of thumb: partitions = (target throughput MB/s) / (single-partition throughput MB/s)
kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --create \
  --topic order-events \
  --partitions 3 \
  --replication-factor 3 \
  --config retention.ms=604800000    # 7 days retention
  --config segment.bytes=536870912   # Roll a new segment file at 512MB

# ── Step 2: Inspect the physical log files on the broker ──
# Each partition gets its own directory: <topic>-<partition-number>
ls -lh /var/kafka-logs/order-events-0/
# Expected output:
# 00000000000000000000.index   <- sparse offset index for fast seek
# 00000000000000000000.log     <- the actual message data, append-only
# 00000000000000000000.timeindex <- maps timestamps to offsets
# leader-epoch-checkpoint      <- tracks leader changes for safe consumer recovery

# ── Step 3: Dump raw messages from a partition segment to inspect offsets ──
# kafka-dump-log is your best friend for debugging corruption or offset gaps
kafka-dump-log.sh \
  --files /var/kafka-logs/order-events-0/00000000000000000000.log \
  --print-data-log | head -20

# ── Step 4: Check topic configuration and current low/high watermarks ──
# High watermark = last offset fully replicated to all ISR members
# Log end offset = last offset written (may not be replicated yet)
kafka-log-dirs.sh \
  --bootstrap-server localhost:9092 \
  --topic-list order-events \
  --describe | python3 -m json.tool

package io.thecodeforge.kafka;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

public class OrderEventConsumer {

    // Shared shutdown flag — set from a shutdown hook so the poll loop exits cleanly
    private final AtomicBoolean shutdownRequested = new AtomicBoolean(false);

    public void startConsuming() {
        Properties consumerConfig = new Properties();

        consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-1:9092,kafka-broker-2:9092");
        consumerConfig.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-service"); // Consumer group name
        consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

        // CRITICAL: How long between poll() calls before the broker declares this consumer dead.
        // If your business logic per batch takes > 5 min, raise this — or better, move processing off the poll thread.
        consumerConfig.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000"); // 5 minutes

        // CRITICAL: How many records to fetch per partition per poll() call.
        // Lower this if your processing is slow to avoid triggering max.poll.interval.ms
        consumerConfig.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100");

        // Use CooperativeStickyAssignor to avoid full stop-the-world rebalances (Kafka 2.4+)
        consumerConfig.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
            CooperativeStickyAssignor.class.getName());

        // Disable auto-commit — we commit manually AFTER processing succeeds
        // Auto-commit is at-most-once: it can commit before your DB write finishes
        consumerConfig.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        // Start reading from the earliest available offset if no committed offset exists
        consumerConfig.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        try (KafkaConsumer<String, String> orderConsumer = new KafkaConsumer<>(consumerConfig)) {

            // Register a ConsumerRebalanceListener to log rebalance events in production
            orderConsumer.subscribe(Collections.singletonList("order-events"),
                new ConsumerRebalanceListener() {
                    @Override
                    public void onPartitionsRevoked(java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {
                        // Called BEFORE partitions are taken away — commit offsets here to avoid reprocessing
                        System.out.printf("[REBALANCE] Revoking partitions: %s — committing offsets now%n", partitions);
                        orderConsumer.commitSync(); // Synchronous commit before giving partitions up
                    }

                    @Override
                    public void onPartitionsAssigned(java.util.Collection<org.apache.kafka.common.TopicPartition> partitions) {
                        System.out.printf("[REBALANCE] Assigned partitions: %s%n", partitions);
                    }
                });

            // Register shutdown hook so Ctrl+C triggers a clean wakeup instead of an abrupt kill
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                shutdownRequested.set(true);
                orderConsumer.wakeup(); // Causes the blocking poll() to throw WakeupException
            }));

            while (!shutdownRequested.get()) {
                try {
                    // poll() does two things: fetches records AND sends heartbeats to the group coordinator
                    ConsumerRecords<String, String> orderBatch = orderConsumer.poll(Duration.ofMillis(500));

                    for (ConsumerRecord<String, String> orderRecord : orderBatch) {
                        System.out.printf("Partition=%d | Offset=%d | Key=%s | Value=%s%n",
                            orderRecord.partition(),
                            orderRecord.offset(),       // Monotonically increasing per partition
                            orderRecord.key(),
                            orderRecord.value());

                        processOrder(orderRecord.value()); // Your actual business logic
                    }

                    if (!orderBatch.isEmpty()) {
                        // Synchronous commit: blocks until broker acknowledges.
                        // Guarantees at-least-once delivery — your processing must be idempotent.
                        orderConsumer.commitSync();
                        System.out.printf("Committed offsets for %d records%n", orderBatch.count());
                    }

                } catch (WakeupException shutdownSignal) {
                    // Expected during shutdown — exit the loop cleanly
                    if (!shutdownRequested.get()) throw shutdownSignal;
                    System.out.println("Shutdown signal received — exiting poll loop");
                }
            }
        }
    }

    private void processOrder(String orderJson) {
        // Simulate order processing — in production this writes to a DB, calls downstream services, etc.
        System.out.printf("Processing order: %s%n", orderJson);
    }

    public static void main(String[] args) {
        new OrderEventConsumer().startConsuming();
    }
}

package io.thecodeforge.kafka;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class DurableOrderEventProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        Properties producerConfig = new Properties();

        producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092");
        producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // ── DURABILITY CONFIG ──────────────────────────────────────────────────
        // Require ALL in-sync replicas to acknowledge before the write is confirmed.
        // Combined with min.insync.replicas=2 on the broker/topic, this means
        // the write survives a single broker failure.
        producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");

        // Enable idempotent producer: assigns this producer a unique PID.
        // Broker deduplicates retried messages using PID + partition + sequence number.
        // REQUIRES: acks=all, retries > 0, max.in.flight.requests.per.connection <= 5
        producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");

        // How many times to retry a failed send before giving up.
        // With idempotence enabled, retries are safe — no duplicates.
        producerConfig.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);

        // Max unacknowledged requests per connection.
        // Must be <= 5 when idempotence is enabled.
        // Higher values increase throughput but with idempotence, 5 is the max safe value.
        producerConfig.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "5");

        // ── THROUGHPUT / LATENCY TUNING ───────────────────────────────────────
        // linger.ms: How long to wait for more messages before sending a batch.
        // 0 = send immediately (low latency, small batches).
        // 20 = wait 20ms, batch more records, fewer network round-trips (higher throughput).
        producerConfig.put(ProducerConfig.LINGER_MS_CONFIG, "20");

        // batch.size: Max bytes per batch per partition before sending regardless of linger.ms.
        // Default is 16KB — increasing to 64KB or 128KB dramatically improves throughput
        // when producing at high volume.
        producerConfig.put(ProducerConfig.BATCH_SIZE_CONFIG, "65536"); // 64KB

        // compression.type: Compresses batches before sending. 'snappy' is CPU-light.
        // 'lz4' is faster. 'zstd' gives best compression ratio (Kafka 2.1+).
        // Compression happens client-side — reduces network and disk usage.
        producerConfig.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        // buffer.memory: Total memory the producer uses to buffer records before sending.
        // If the buffer fills up (broker is slow), send() will block for max.block.ms,
        // then throw a TimeoutException.
        producerConfig.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "67108864"); // 64MB

        try (KafkaProducer<String, String> orderProducer = new KafkaProducer<>(producerConfig)) {

            String orderId = "orderId-9821";
            String orderPayload = "{\"status\":\"PLACED\",\"amount\":149.99,\"customerId\":\"cust-4477\"}";

            // Using the orderId as the message key ensures all events for the same order
            // always land on the same partition — preserving per-order event ordering.
            ProducerRecord<String, String> orderRecord = new ProducerRecord<>(
                "order-events",   // topic
                orderId,          // key — determines partition via murmur2 hash
                orderPayload      // value
            );

            // ── OPTION A: Async send with callback (preferred for high throughput) ──
            orderProducer.send(orderRecord, (metadata, sendException) -> {
                if (sendException != null) {
                    // In production: push to a dead-letter topic or alert on-call
                    System.err.printf("FAILED to send order %s: %s%n", orderId, sendException.getMessage());
                } else {
                    System.out.printf("[ASYNC] Sent to topic=%s partition=%d offset=%d timestamp=%d%n",
                        metadata.topic(),
                        metadata.partition(),
                        metadata.offset(),
                        metadata.timestamp());
                }
            });

            // ── OPTION B: Sync send — blocks until ack or exception ──
            // Use when you MUST know the write succeeded before proceeding (e.g., financial transactions)
            try {
                RecordMetadata syncMetadata = orderProducer.send(orderRecord).get(); // Blocks here
                System.out.printf("[SYNC] Confirmed at partition=%d offset=%d%n",
                    syncMetadata.partition(),
                    syncMetadata.offset());
            } catch (ExecutionException kafkaSendError) {
                System.err.println("[SYNC] Send failed: " + kafkaSendError.getCause().getMessage());
            }

            // flush() ensures all buffered messages are sent before the try-with-resources closes the producer
            orderProducer.flush();
        }
    }
}

# ── Check ISR status for all partitions of a topic ──
# Under-replicated partitions (ISR < replication factor) are a red alert
kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --topic order-events

# Expected healthy output:
# Topic: order-events  Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2,3
# Topic: order-events  Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,3,1
# Topic: order-events  Partition: 2  Leader: 3  Replicas: 3,1,2  Isr: 3,1,2

# ── Simulate broker 2 going down, then check ISR shrinkage ──
# After replica.lag.time.max.ms (30s), broker 2 is removed from ISR
kafka-topics.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --topic order-events

# Output after broker 2 failure:
# Topic: order-events  Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,3  <-- broker 2 dropped!
# Topic: order-events  Partition: 1  Leader: 3  Replicas: 2,3,1  Isr: 3,1  <-- leader elected from ISR
# Topic: order-events  Partition: 2  Leader: 3  Replicas: 3,1,2  Isr: 3,1

# ── Check consumer group lag — the most important production health metric ──
# LAG = LOG-END-OFFSET minus CURRENT-OFFSET per partition
# Growing lag means consumers can't keep up with producer throughput
kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group order-processing-service

# Healthy output (lag near zero):
# GROUP                    TOPIC        PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID
# order-processing-service order-events 0          1047            1047            0    consumer-1-uuid
# order-processing-service order-events 1          892             892             0    consumer-2-uuid
# order-processing-service order-events 2          1103            1103            0    consumer-1-uuid

# ── Set min.insync.replicas at the topic level (overrides broker default) ──
# This is the safest place to set it — per topic, not globally
kafka-configs.sh \
  --bootstrap-server localhost:9092 \
  --alter \
  --entity-type topics \
  --entity-name order-events \
  --add-config min.insync.replicas=2

# ── Verify the config was applied ──
kafka-configs.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --entity-type topics \
  --entity-name order-events