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

// io.thecodeforge — database tutorial

// Check the current controller broker
SELECT * FROM kafka_broker_roles WHERE role = 'controller';
-- Output: broker_id=2, host=kafka-prod-02.internal

// View partition leader distribution
SELECT topic, partition, leader, replicas, isr
FROM kafka_partition_state
WHERE topic = 'order_events';

// Sample output for a partition with a recent leader change
-- topic: order_events
-- partition: 7
-- leader: 3
-- replicas: [1,2,3]
-- isr: [2,3]
// Notice leader=3 but broker 1 is out of sync. Leader election picked 3 from ISR.

// Tune controller failover detection (server.properties)
-- zookeeper.session.timeout.ms=6000
-- zookeeper.connection.timeout.ms=18000
-- controlled.shutdown.enable=true

// io.thecodeforge — database tutorial

// Query consumer group state from Kafka's internal __consumer_offsets topic
SELECT group_id, state, partition, current_offset, log_end_offset
FROM kafka_group_state
WHERE group_id = 'order_processor_v2';

-- Example state: 'Stable' or 'PreparingRebalance'
// If state is 'PreparingRebalance', consumers are NOT processing.

// Check rebalance rate in the last 5 minutes (JMX metric)
-- Metric: kafka.consumer:type=consumer-coordinator-metrics
-- Attribute: rebalance-rate
-- Value: 2.3 (rebalances per minute) => BAD

// Sample output:
-- group_id: order_processor_v2
-- state: Stable
-- partition: 4
-- current_offset: 1423508
-- log_end_offset: 1423512
// This partition is nearly caught up. Good.

// But if state flips to PreparingRebalance, offsets stall.

// io.thecodeforge — database tutorial

// Create a topic with 3 partitions, replication factor 2
kafka-topics \
  --bootstrap-server broker1:9092 \
  --create \
  --topic orders_raw \
  --partitions 3 \
  --replication-factor 2

// Produce a single record with a key for co-partitioning
echo "order_id:abc123,event:created" | \
  kafka-console-producer \
  --bootstrap-server broker1:9092 \
  --topic orders_raw \
  --property parse.key=true \
  --property key.separator=:

// Describe consumer lag to find slow partitions
kafka-consumer-groups \
  --bootstrap-server broker1:9092 \
  --group order_processor \
  --describe

// io.thecodeforge — database tutorial

// Check per-broker partition distribution
kafka-topics \
  --bootstrap-server broker1:9092 \
  --describe \
  --topic orders_raw

// Output shows partition-to-broker mapping
Topic: orders_raw  PartitionCount: 3  ReplicationFactor: 2
  Topic: orders_raw  Partition: 0  Leader: 1  Replicas: 1,2  Isr: 1,2
  Topic: orders_raw  Partition: 1  Leader: 2  Replicas: 2,3  Isr: 2,3
  Topic: orders_raw  Partition: 2  Leader: 3  Replicas: 3,1  Isr: 3,1

// Simulate adding 3 more partitions (DANGER if keyed!)
kafka-topics \
  --bootstrap-server broker1:9092 \
  --alter \
  --topic orders_raw \
  --partition-count 6

// io.thecodeforge — database tutorial

// Producer config for log aggregation (batch-heavy)
producer:
  bootstrap.servers: "broker1:9092,broker2:9092"
  acks: "1"                  # tolerate rare loss for throughput
  batch.size: 65536          # 64KB batches for small log messages
  linger.ms: 100             # wait up to 100ms to fill a batch
  compression.type: "gzip"  # logs compress 10:1 on text
  key.serializer: "org.apache.kafka.common.serialization.StringSerializer"
  value.serializer: "org.apache.kafka.common.serialization.StringSerializer"

// Single log record (keyed by service name for order per service)
Key: "auth-service"
Value: "{\"ts\":\"2025-03-15T10:30:00Z\",\"level\":\"ERROR\",\"service\":\"auth-service\",\"trace_id\":\"xyz789\",\"message\":\"DynamoDB timeout on user lookup\"}"

// io.thecodeforge — database tutorial

-- Simulated Kafka order lifecycle stream
CREATE STREAM orders (
  order_id STRING KEY,
  user_id STRING,
  product_id STRING,
  amount DOUBLE,
  status STRING
) WITH (
  kafka_topic = 'orders',
  value_format = 'JSON',
  partitions = 6
);

-- Consumer: fraud detection
SELECT order_id, user_id, amount
FROM orders
WHERE status = 'PENDING'
  AND amount > 10000
EMIT CHANGES;

// io.thecodeforge — database tutorial

-- Kafka transactional producer setup (conceptual SQL)
CREATE PRODUCER transaction_producer WITH (
  transactional.id = 'order-payment-xact',
  enable.idempotence = true
);

BEGIN TRANSACTION;
INSERT INTO orders (order_id, status) VALUES ('O123', 'PAID');
INSERT INTO payments (order_id, amount) VALUES ('O123', 49.99);
COMMIT;

// io.thecodeforge — database tutorial

CREATE STREAM user_activity (
  user_id STRING KEY,
  event_type STRING,
  page STRING,
  timestamp BIGINT
) WITH (
  kafka_topic = 'user_activity',
  partitions = 12,
  value_format = 'AVRO'
);

-- Realtime active user count
SELECT user_id, COUNT(*) AS actions
FROM user_activity
WINDOW TUMBLING (SIZE 5 MINUTES)
GROUP BY user_id
HAVING COUNT(*) > 10
EMIT CHANGES;

// io.thecodeforge — database tutorial
// 25 lines max
#!/bin/bash
# macOS
brew install kafka
brew services start zookeeper
brew services start kafka
# Linux/WSL2
wget https://dlcdn.apache.org/kafka/3.5.1/kafka_2.13-3.5.1.tgz
tar -xzf kafka_2.13-3.5.1.tgz
cd kafka_2.13-3.5.1
bin/zookeeper-server-start.sh config/zookeeper.properties &
bin/kafka-server-start.sh config/server.properties &
echo "Verify: bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092"

// io.thecodeforge — database tutorial
// 25 lines max
-- Terminal 1: ZooKeeper first
!ZooKeeper on WSL2
bin/zookeeper-server-start.sh config/zookeeper.properties
-- Terminal 2: Kafka broker
bin/kafka-server-start.sh config/server.properties
-- Verify
bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
-- Create topic
bin/kafka-topics.sh --create --topic orders --partitions 3 --replication-factor 1 --bootstrap-server localhost:9092
-- Status
bin/kafka-topics.sh --describe --topic orders --bootstrap-server localhost:9092

// io.thecodeforge — database tutorial
// 25 lines max
-- Produce with key
bin/kafka-console-producer.sh --topic orders --bootstrap-server localhost:9092 --property parse.key=true --property key.separator=:
>order:123:confirmed
-- Consume in group
bin/kafka-console-consumer.sh --topic orders --group my-group --bootstrap-server localhost:9092 --from-beginning --property print.key=true --property print.partition
-- Describe group
bin/kafka-consumer-groups.sh --group my-group --describe --bootstrap-server localhost:9092
GROUP           TOPIC  PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
my-group        orders 0          1               2              1
my-group        orders 1          0               1              1