Kafka Consumer Deep Dive

                         Topic: orders (6 partitions)
                    ┌─────┬─────┬─────┬─────┬─────┬─────┐
                    │ P0  │ P1  │ P2  │ P3  │ P4  │ P5  │
                    └──┬──┴──┬──┴──┬──┴──┬──┴──┬──┴──┬──┘
                       │     │     │     │     │     │
           ┌───────────┴─────┴─────┴─────┴─────┴─────┘
           │         Consumer Group: order-processors
           │
    ┌──────┴──────┬─────────────┬─────────────┐
    ▼             ▼             ▼             ▼
┌────────┐   ┌────────┐   ┌────────┐   ┌────────┐
│Consumer│   │Consumer│   │Consumer│   │Consumer│
│   A    │   │   B    │   │   C    │   │   D    │
│ P0,P1  │   │  P2,P3 │   │   P4   │   │   P5   │
└────────┘   └────────┘   └────────┘   └────────┘

┌─────────────────────────────────────────────────────────────┐
│                  __consumer_offsets (50 partitions)         │
│  ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┬─────┬───────┐  │
│  │ P0  │ P1  │ P2  │ ... │ P25 │ ... │ P47 │ P48 │  P49  │  │
│  └──┬──┴─────┴─────┴─────┴──┬──┴─────┴─────┴─────┴───────┘  │
│     │                       │                                │
│  Broker1               Broker3                               │
│  (Leader P0)           (Leader P25)                          │
│     │                       │                                │
│  Coordinates:           Coordinates:                         │
│  • group-A              • group-B                            │
│  • group-C              • group-D                            │
└─────────────────────────────────────────────────────────────┘

Consumer A          Coordinator          Consumer B
    │                    │                    │
    │───JoinGroup───────→│←────JoinGroup──────│
    │                    │                    │
    │   (waits for all members to join)       │
    │                    │                    │
    │←──JoinResponse─────│────JoinResponse───→│
    │   (A elected       │                    │
    │    leader)         │                    │
    │                    │                    │
    │───SyncGroup────────│←────SyncGroup──────│
    │   (with            │   (empty)          │
    │    assignments)    │                    │
    │                    │                    │
    │←──SyncResponse─────│────SyncResponse───→│
    │   (P0,P1)          │   (P2,P3)          │

Consumer A          Coordinator          Consumer B
    │                    │                    │
    │──Heartbeat(sub)───→│                    │
    │                    │←──Heartbeat(sub)───│
    │                    │                    │
    │    (Coordinator computes assignment)    │
    │                    │                    │
    │←─Heartbeat(assign)─│                    │
    │   (P0,P1,P2)       │                    │
    │                    │                    │
    │                    │──Heartbeat(assign)→│
    │                    │   (P3,P4,P5)       │
    │                    │                    │
    │   (Consumers reconcile incrementally)   │

var config = Map.of(
    "group.protocol"       , "consumer", // Enable KIP-848
    "group.remote.assignor", "uniform"   // Server-side assignor (uniform/range)
    // Other standard configs...
);

// Classic protocol with cooperative rebalancing
var config = Map.of(
    "partition.assignment.strategy", "org.apache.kafka.clients.consumer.CooperativeStickyAssignor"
);

Key: (group_id, topic, partition)
Value: (offset, leader_epoch, metadata, timestamp)

var config = Map.of(
    "enable.auto.commit"     , "true",
    "auto.commit.interval.ms", "5000"
);
// Danger: Offsets committed before processing completes

while (true) {
    var records = consumer.poll(Duration.ofMillis(100));
    for (var record : records) {
        process(record);
    }
    consumer.commitSync(); // Blocks until confirmed
}

while (true) {
    var records = consumer.poll(Duration.ofMillis(100));
    for (var record : records) {
        process(record);
    }
    consumer.commitAsync((offsets, exception) -> {
        if (exception != null) {
            logger.warn("Commit failed: " + exception.getMessage());
        }
    });
}

int processed = 0;
var currentOffsets = new HashMap<TopicPartition, OffsetAndMetadata>();
 
while (true) {
    var records = consumer.poll(Duration.ofMillis(100));
    for (var record : records) {
        process(record);
        currentOffsets.put(new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1));
        processed++;
 
        if (processed % 100 == 0) {
            consumer.commitSync(currentOffsets);
            currentOffsets.clear();
        }
    }
}

poll() returns 1000 messages
     │
     ▼
Auto-commit fires (offset = 1000)  ← Kafka thinks you're done
     │
     ▼
Processing message 500...
     │
     ▼
💥 Application crashes
     │
     ▼
Restart: consumer resumes at offset 1001
     │
     ▼
Messages 501-1000 are LOST

// Reset to 24 hours ago instead of earliest/latest
var config = Map.of(
    "auto.offset.reset", "by_duration:PT24H" // ISO 8601 duration
);

var config = Map.of(
    "isolation.level", "read_committed" // Only see committed transactional messages
    // Default: "read_uncommitted"
);

try (var consumer = new KafkaConsumer<String, String>(config)) {
    consumer.subscribe(List.of("orders"));
 
    while (running.get()) {
        var records = consumer.poll(Duration.ofMillis(100));
        for (var record : records) {
            processRecord(record);
        }
    }
}

var records = consumer.poll(Duration.ofMillis(100));
consumer.pause(consumer.assignment()); // Stop fetching
 
for (var record : records) {
    processSlowly(record); // Can take time
}
 
consumer.resume(consumer.assignment()); // Resume fetching

var config = Map.of(
    "max.poll.interval.ms", "600000", // 10 minutes
    "max.poll.records"     , "100"    // Smaller batches
);

var executor = Executors.newFixedThreadPool(10);
 
while (running.get()) {
    var records = consumer.poll(Duration.ofMillis(100));
    var futures = new ArrayList<Future<?>>();
    for (var record : records) {
        futures.add(executor.submit(() -> process(record)));
    }
    // Wait for batch, then commit
    for (var future : futures) {
        future.get();
    }
    consumer.commitSync();
}

int retries = 0;
while (retries < 3) {
    try {
        process(record);
        retries = 3; // Success
    } catch (Exception e) {
        retries++;
        Thread.sleep(100L * retries);
        if (retries == 3) throw e; // Give up
    }
}

try {
    process(record);
} catch (Exception e) {
    logger.error("Failed to process, sending to DLQ", e);
    producer.send(new ProducerRecord<>("my-app-dlq", record.key(), record.value()));
    // Functionally "consumed" even though processing failed
}

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        logger.info("Revoked: " + partitions);
        // Commit current progress before losing partitions
        consumer.commitSync();
        // Flush any in-memory state
        stateStore.flush();
    }
 
    @Override
    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        logger.info("Assigned: " + partitions);
        // Initialize state for new partitions
        for (var tp : partitions) {
            stateStore.initialize(tp);
        }
    }
});

var config = Map.of(
    "group.instance.id", "order-processor-" + hostname
);

var config = Map.of(
    "client.rack", "us-east-1a" // The zone this consumer is running in
    // Brokers must also have broker.rack configured
);

Lag = Log End Offset - Consumer Committed Offset

kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --describe \
  --group order-processors
 
# Output:
# GROUP            TOPIC   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# order-processors orders  0          15234           15240           6
# order-processors orders  1          14892           14892           0