10.1 Practical Consumer Patterns & Real-World Scenarios

Module Overview

Goal: Compose all previous knowledge — no new theory, only real-world application.

What You'll Learn:

• Batch vs streaming processing patterns
• Thread-per-partition model
• Graceful shutdown strategies
• Backpressure-aware consumers
• Exactly-once pipelines (consumer role)
• Common anti-patterns and solutions
• Production-ready consumer templates

Assumes: All previous modules (this is the capstone).

Key Insight: Real systems combine multiple patterns. This module shows you how.

1. Pattern 1: High-Throughput Batch Consumer

Use Case

Scenario: Process millions of events/hour, batch write to database.

Requirements:

• Maximize throughput
• Batch database inserts
• At-least-once delivery OK (idempotent)

Implementation

public class BatchConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final DatabaseWriter db;
    private final int batchSize = 1000;
    
    public BatchConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "batch-processor");
        props.put("enable.auto.commit", "false");
        
        // Throughput-optimized
        props.put("max.poll.records", 5000);
        props.put("fetch.min.bytes", 1048576); // 1 MB
        props.put("fetch.max.wait.ms", 500);
        
        this.consumer = new KafkaConsumer<>(props);
        this.db = new DatabaseWriter();
    }
    
    public void run() {
        consumer.subscribe(Collections.singletonList("events"));
        
        List<Record> batch = new ArrayList<>();
        
        while (running) {
            ConsumerRecords<String, String> records = 
                consumer.poll(Duration.ofMillis(100));
            
            for (ConsumerRecord<String, String> record : records) {
                batch.add(parseRecord(record));
                
                if (batch.size() >= batchSize) {
                    db.batchInsert(batch);
                    batch.clear();
                    consumer.commitSync();
                }
            }
        }
        
        // Final batch
        if (!batch.isEmpty()) {
            db.batchInsert(batch);
            consumer.commitSync();
        }
        
        consumer.close();
    }
}

Expected Performance:

• 100,000+ records/sec
• Database inserts every 1-2 seconds
• Low overhead

2. Pattern 2: Low-Latency Stream Processor

Use Case

Scenario: Real-time fraud detection, sub-second latency.

Requirements:

• Minimize latency
• Process each message immediately
• Fast failure detection

Implementation

public class LowLatencyConsumer {
    private final KafkaConsumer<String, Transaction> consumer;
    private final FraudDetector fraudDetector;
    
    public LowLatencyConsumer() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "fraud-detection");
        props.put("enable.auto.commit", "false");
        
        // Latency-optimized
        props.put("max.poll.records", 50);
        props.put("fetch.min.bytes", 1);
        props.put("fetch.max.wait.ms", 0);
        
        // Fast failover
        props.put("session.timeout.ms", 6000);
        props.put("heartbeat.interval.ms", 2000);
        
        this.consumer = new KafkaConsumer<>(props);
        this.fraudDetector = new FraudDetector();
    }
    
    public void run() {
        consumer.subscribe(Collections.singletonList("transactions"));
        
        while (running) {
            ConsumerRecords<String, Transaction> records = 
                consumer.poll(Duration.ofMillis(100));
            
            for (ConsumerRecord<String, Transaction> record : records) {
                long startTime = System.nanoTime();
                
                // Process immediately
                FraudResult result = fraudDetector.analyze(record.value());
                
                if (result.isFraud()) {
                    alertService.send(result);
                }
                
                long latency = (System.nanoTime() - startTime) / 1_000_000;
                metrics.recordLatency(latency);
            }
            
            // Commit per batch (fast loop)
            consumer.commitAsync();
        }
        
        consumer.commitSync();
        consumer.close();
    }
}

Expected Performance:

• < 10ms p99 latency
• 10,000+ transactions/sec
• Fast fraud detection

3. Pattern 3: Idempotent Consumer

Use Case

Scenario: At-least-once delivery, but need idempotent processing.

Requirements:

• No duplicate processing
• Deduplication mechanism
• Crash recovery

Implementation

public class IdempotentConsumer {
    private final KafkaConsumer<String, Order> consumer;
    private final OrderProcessor processor;
    private final Set<String> processedIds;
    
    public IdempotentConsumer() {
        this.consumer = new KafkaConsumer<>(props);
        this.processor = new OrderProcessor();
        this.processedIds = loadProcessedIds(); // From DB or cache
    }
    
    public void run() {
        consumer.subscribe(Collections.singletonList("orders"));
        
        while (running) {
            ConsumerRecords<String, Order> records = 
                consumer.poll(Duration.ofMillis(100));
            
            for (ConsumerRecord<String, Order> record : records) {
                String orderId = record.key();
                
                // Idempotence check
                if (processedIds.contains(orderId)) {
                    logger.info("Skipping duplicate order: {}", orderId);
                    continue;
                }
                
                // Process
                processor.process(record.value());
                
                // Mark as processed (in DB transaction with business logic)
                database.transaction(() -> {
                    processor.save(record.value());
                    database.markProcessed(orderId);
                });
                
                processedIds.add(orderId);
            }
            
            consumer.commitSync();
        }
        
        consumer.close();
    }
}

4. Pattern 4: Thread-Per-Partition Model

Use Case

Scenario: Need parallelism with ordering guarantees.

Requirements:

• Parallel processing
• Maintain partition ordering
• Scale with partition count

Implementation

public class ThreadPerPartitionConsumer {
    private final ExecutorService executor;
    private final Map<Integer, BlockingQueue<ConsumerRecord>> partitionQueues;
    
    public ThreadPerPartitionConsumer(int partitionCount) {
        this.executor = Executors.newFixedThreadPool(partitionCount);
        this.partitionQueues = new ConcurrentHashMap<>();
        
        // Create queue and worker thread per partition
        for (int i = 0; i < partitionCount; i++) {
            int partition = i;
            BlockingQueue<ConsumerRecord> queue = new LinkedBlockingQueue<>(1000);
            partitionQueues.put(partition, queue);
            
            // Worker thread for this partition
            executor.submit(() -> processPartition(partition, queue));
        }
    }
    
    public void run() {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("orders"));
        
        while (running) {
            ConsumerRecords<String, String> records = 
                consumer.poll(Duration.ofMillis(100));
            
            // Distribute to partition queues
            for (ConsumerRecord<String, String> record : records) {
                partitionQueues.get(record.partition()).offer(record);
            }
            
            consumer.commitSync();
        }
        
        executor.shutdown();
        consumer.close();
    }
    
    private void processPartition(int partition, BlockingQueue<ConsumerRecord> queue) {
        while (running) {
            try {
                ConsumerRecord record = queue.poll(100, TimeUnit.MILLISECONDS);
                if (record != null) {
                    processRecord(record);
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
}

Benefits:

• Parallel processing per partition
• Ordering maintained within partition
• Scales with partition count

Pattern 4b: Work Queue with External Offset Tracking

Advanced Variation: Decouple consuming from processing with a work queue, but manage offset commits carefully.

The Challenge

Consumer Thread (fast):
├─ poll() records from Kafka
└─ put() records into BlockingQueue

Worker Threads (slow):
├─ take() records from BlockingQueue
└─ process records in parallel

Problem: When to commit offsets?
- Can't commit immediately after poll() (records not processed yet)
- Can't let workers commit (they don't know about ordering)
- Must track which offsets are safe to commit

Implementation: Safe Offset Tracking

public class WorkQueueConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final BlockingQueue<ConsumerRecord<String, String>> workQueue;
    private final ExecutorService workers;
    
    // Track in-flight offsets per partition
    private final Map<TopicPartition, OffsetTracker> offsetTrackers;
    
    public WorkQueueConsumer(int workerCount) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "work-queue-processor");
        props.put("enable.auto.commit", "false");
        props.put("max.poll.records", 500);
        
        this.consumer = new KafkaConsumer<>(props);
        this.workQueue = new LinkedBlockingQueue<>(10000);
        this.workers = Executors.newFixedThreadPool(workerCount);
        this.offsetTrackers = new ConcurrentHashMap<>();
        
        // Start worker threads
        for (int i = 0; i < workerCount; i++) {
            workers.submit(this::workerLoop);
        }
    }
    
    // Consumer thread: Poll and enqueue
    public void run() {
        consumer.subscribe(Collections.singletonList("orders"));
        
        while (running.get()) {
            ConsumerRecords<String, String> records = 
                consumer.poll(Duration.ofMillis(100));
            
            // Enqueue all records
            for (ConsumerRecord<String, String> record : records) {
                TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                
                // Track that this offset is in-flight
                offsetTrackers
                    .computeIfAbsent(tp, k -> new OffsetTracker())
                    .markInFlight(record.offset());
                
                try {
                    // Block if queue is full (backpressure)
                    workQueue.put(record);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
            
            // Commit safe offsets (all previous offsets processed)
            commitSafeOffsets();
        }
        
        // Graceful shutdown
        workers.shutdown();
        workers.awaitTermination(30, TimeUnit.SECONDS);
        commitSafeOffsets(); // Final commit
        consumer.close();
    }
    
    // Worker thread: Process records
    private void workerLoop() {
        while (running.get() || !workQueue.isEmpty()) {
            try {
                ConsumerRecord<String, String> record = 
                    workQueue.poll(100, TimeUnit.MILLISECONDS);
                
                if (record != null) {
                    try {
                        // Process record
                        processRecord(record);
                        
                        // Mark offset as processed
                        TopicPartition tp = new TopicPartition(
                            record.topic(), 
                            record.partition()
                        );
                        offsetTrackers.get(tp).markProcessed(record.offset());
                        
                    } catch (Exception e) {
                        logger.error("Processing failed for offset {}: {}", 
                            record.offset(), e.getMessage());
                        // Handle error (DLQ, retry, etc.)
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
    
    // Commit only offsets where all previous offsets are processed
    private void commitSafeOffsets() {
        Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>();
        
        for (Map.Entry<TopicPartition, OffsetTracker> entry : offsetTrackers.entrySet()) {
            TopicPartition tp = entry.getKey();
            OffsetTracker tracker = entry.getValue();
            
            // Get highest consecutive processed offset
            long safeOffset = tracker.getHighestConsecutiveProcessed();
            
            if (safeOffset >= 0) {
                offsetsToCommit.put(tp, new OffsetAndMetadata(safeOffset + 1));
            }
        }
        
        if (!offsetsToCommit.isEmpty()) {
            try {
                consumer.commitSync(offsetsToCommit);
            } catch (CommitFailedException e) {
                logger.error("Commit failed: {}", e.getMessage());
            }
        }
    }
    
    // Track which offsets are in-flight vs processed
    private static class OffsetTracker {
        private final Set<Long> inFlight = ConcurrentHashMap.newKeySet();
        private final Set<Long> processed = ConcurrentHashMap.newKeySet();
        private volatile long highestInFlight = -1;
        
        public synchronized void markInFlight(long offset) {
            inFlight.add(offset);
            if (offset > highestInFlight) {
                highestInFlight = offset;
            }
        }
        
        public synchronized void markProcessed(long offset) {
            inFlight.remove(offset);
            processed.add(offset);
        }
        
        // Find highest offset where all previous offsets are processed
        public synchronized long getHighestConsecutiveProcessed() {
            long offset = 0;
            while (processed.contains(offset)) {
                offset++;
            }
            return offset - 1; // Last consecutive processed offset
        }
    }
}

Why This Pattern is Complex

Problem 1: Out-of-Order Completion

Consumer polls: offsets 100, 101, 102, 103, 104
Workers process out of order:

Time  Completed Offsets    Safe to Commit?
────┼───────────────────┼────────────────────
  1 │ 102                │ NO (100, 101 not done)
  2 │ 102, 104           │ NO (100, 101, 103 not done)
  3 │ 102, 104, 100      │ Only 100 (101 not done)
  4 │ 102, 104, 100, 101 │ 100, 101, 102 (103 not done)
  5 │ All                │ 100-104 (all done)

Key Insight: Can only commit offset N if all offsets < N are processed.

Problem 2: Failure During In-Flight

Scenario:
1. Consumer polls offsets 100-109
2. Enqueues all to work queue
3. Workers start processing
4. Consumer crashes (offsets NOT committed)
5. Workers finish processing 100-105 (too late!)
6. Consumer restarts, reads from last committed offset

Result:
- Offsets 100-105 processed twice
- Can't avoid with work queue pattern
- MUST use idempotent processing

Commit Strategy Comparison

When to Use Work Queue Pattern

Use When:

• ✅ Processing is CPU-intensive or slow
• ✅ Want to maximize throughput with parallelism
• ✅ Can tolerate duplicates (idempotent processing)
• ✅ Need backpressure control

Avoid When:

• ❌ Processing is already fast (adds overhead)
• ❌ Cannot implement idempotent processing
• ❌ Ordering across partitions required
• ❌ Simpler patterns suffice

Simplified Alternative: Partition-Based Queues

Instead of one global queue, use one queue per partition:

// One queue per partition = ordering preserved
Map<Integer, BlockingQueue<ConsumerRecord>> partitionQueues = new HashMap<>();
Map<Integer, CompletableFuture<Void>> partitionWorkers = new HashMap<>();

for (TopicPartition tp : consumer.assignment()) {
    int partition = tp.partition();
    BlockingQueue<ConsumerRecord> queue = new LinkedBlockingQueue<>(1000);
    partitionQueues.put(partition, queue);
    
    // One worker per partition
    partitionWorkers.put(partition, CompletableFuture.runAsync(() -> {
        while (running.get()) {
            ConsumerRecord record = queue.poll(100, TimeUnit.MILLISECONDS);
            if (record != null) {
                processRecord(record);
            }
        }
    }));
}

// In poll loop
for (ConsumerRecord record : records) {
    partitionQueues.get(record.partition()).put(record);
}

// After all in-flight processed
consumer.commitSync();

Benefits:

• ✅ Ordering within partition preserved
• ✅ Simpler offset tracking (process partition sequentially)
• ✅ No out-of-order completion issues

Trade-offs:

• ⚠️ Less parallelism (limited to partition count)
• ⚠️ Uneven partition load can bottleneck

Best Practices for Work Queue Pattern

1. Always Use Idempotent Processing

   void processRecord(ConsumerRecord record) {
       String id = extractId(record);
       if (db.isProcessed(id)) {
           return; // Skip duplicate
       }
       // Process and mark as processed in transaction
       db.transaction(() -> {
           doBusinessLogic(record);
           db.markProcessed(id);
       });
   }
   

2. Monitor Queue Depth

   metrics.gauge("work_queue_depth", workQueue.size());
   metrics.gauge("work_queue_capacity", workQueue.remainingCapacity());
   
   // Alert if queue is filling up
   if (workQueue.size() > 0.8 * maxQueueSize) {
       logger.warn("Work queue is {}% full", 
           (100 * workQueue.size() / maxQueueSize));
   }
   

3. Implement Backpressure

   if (workQueue.remainingCapacity() < 100) {
       // Pause Kafka consumption
       consumer.pause(consumer.assignment());
   } else if (workQueue.size() < maxQueueSize * 0.5) {
       // Resume Kafka consumption
       consumer.resume(consumer.assignment());
   }
   

4. Handle Worker Failures

   private void workerLoop() {
       while (running.get()) {
           try {
               ConsumerRecord record = workQueue.poll(100, TimeUnit.MILLISECONDS);
               if (record != null) {
                   retryWithBackoff(() -> processRecord(record));
                   markProcessed(record);
               }
           } catch (Exception e) {
               logger.error("Worker failed", e);
               // Send to DLQ, or retry later
           }
       }
   }
   

Performance Characteristics

Typical Setup:

• 1 consumer thread
• 10 worker threads
• 10,000 record queue

Expected Performance:

Decision Matrix: Consumer Concurrency Patterns

Recommendation:

• Start simple: Single-threaded consumer
• Need more throughput + ordering: Thread-per-partition
• Need maximum throughput + can handle duplicates: Work queue with idempotent processing

5. Pattern 5: Graceful Shutdown

Use Case

Scenario: Safe shutdown without data loss.

Requirements:

• Process in-flight records
• Commit final offsets
• Clean resource cleanup

Implementation

public class GracefulConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final AtomicBoolean running = new AtomicBoolean(true);
    private final CountDownLatch shutdownLatch = new CountDownLatch(1);
    
    public GracefulConsumer() {
        this.consumer = new KafkaConsumer<>(props);
        
        // Register shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            logger.info("Shutdown signal received");
            running.set(false);
            consumer.wakeup();
            
            try {
                shutdownLatch.await(30, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                logger.error("Shutdown interrupted", e);
            }
        }));
    }
    
    public void run() {
        try {
            consumer.subscribe(
                Collections.singletonList("orders"),
                new ConsumerRebalanceListener() {
                    @Override
                    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                        logger.info("Partitions revoked, committing offsets");
                        consumer.commitSync();
                    }
                    
                    @Override
                    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                        logger.info("Partitions assigned: {}", partitions);
                    }
                }
            );
            
            while (running.get()) {
                ConsumerRecords<String, String> records = 
                    consumer.poll(Duration.ofMillis(100));
                
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(record);
                }
                
                consumer.commitSync();
            }
            
        } catch (WakeupException e) {
            // Expected during shutdown
            if (running.get()) {
                throw e; // Unexpected wakeup
            }
        } finally {
            try {
                // Final commit
                consumer.commitSync();
                logger.info("Final commit completed");
            } finally {
                consumer.close();
                logger.info("Consumer closed");
                shutdownLatch.countDown();
            }
        }
    }
}

6. Pattern 6: Backpressure-Aware Consumer

Use Case

Scenario: Consumer faster than downstream system.

Requirements:

• Adapt to downstream capacity
• Avoid overwhelming downstream
• Maintain stability

Implementation

public class BackpressureConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final DownstreamService downstream;
    private final BlockingQueue<Record> buffer;
    private final int maxBufferSize = 10000;
    
    public BackpressureConsumer() {
        this.consumer = new KafkaConsumer<>(props);
        this.downstream = new DownstreamService();
        this.buffer = new LinkedBlockingQueue<>(maxBufferSize);
        
        // Start processing thread
        new Thread(this::processBuffer).start();
    }
    
    public void run() {
        consumer.subscribe(Collections.singletonList("events"));
        
        while (running) {
            // Check buffer capacity
            int availableCapacity = maxBufferSize - buffer.size();
            
            if (availableCapacity < maxBufferSize * 0.1) { // < 10% capacity
                // Pause consumption
                consumer.pause(consumer.assignment());
                logger.warn("Buffer full, pausing consumption");
            } else if (availableCapacity > maxBufferSize * 0.5) { // > 50% capacity
                // Resume consumption
                consumer.resume(consumer.assignment());
            }
            
            ConsumerRecords<String, String> records = 
                consumer.poll(Duration.ofMillis(100));
            
            for (ConsumerRecord<String, String> record : records) {
                buffer.offer(record);
            }
            
            consumer.commitAsync();
        }
        
        consumer.close();
    }
    
    private void processBuffer() {
        while (running) {
            try {
                Record record = buffer.poll(100, TimeUnit.MILLISECONDS);
                if (record != null) {
                    downstream.send(record); // May be slow
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }
}

7. Pattern 7: Exactly-Once Consumer-Producer

Use Case

Scenario: Consume from one topic, transform, produce to another (exactly-once).

Requirements:

• No duplicates in output
• No data loss
• Transactional semantics

Implementation

public class ExactlyOnceProcessor {
    private final KafkaConsumer<String, String> consumer;
    private final KafkaProducer<String, String> producer;
    
    public ExactlyOnceProcessor() {
        // Consumer config
        Properties consumerProps = new Properties();
        consumerProps.put("bootstrap.servers", "localhost:9092");
        consumerProps.put("group.id", "exactly-once-processor");
        consumerProps.put("enable.auto.commit", "false");
        consumerProps.put("isolation.level", "read_committed");
        
        this.consumer = new KafkaConsumer<>(consumerProps);
        
        // Producer config (transactional)
        Properties producerProps = new Properties();
        producerProps.put("bootstrap.servers", "localhost:9092");
        producerProps.put("transactional.id", "processor-1");
        producerProps.put("enable.idempotence", "true");
        
        this.producer = new KafkaProducer<>(producerProps);
        producer.initTransactions();
    }
    
    public void run() {
        consumer.subscribe(Collections.singletonList("input-topic"));
        
        while (running) {
            ConsumerRecords<String, String> records = 
                consumer.poll(Duration.ofMillis(100));
            
            if (records.isEmpty()) {
                continue;
            }
            
            producer.beginTransaction();
            
            try {
                for (ConsumerRecord<String, String> record : records) {
                    // Transform
                    String transformed = transform(record.value());
                    
                    // Produce
                    producer.send(new ProducerRecord<>(
                        "output-topic",
                        record.key(),
                        transformed
                    ));
                }
                
                // Commit offsets to transaction
                Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
                for (TopicPartition partition : records.partitions()) {
                    List<ConsumerRecord<String, String>> partitionRecords = 
                        records.records(partition);
                    long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
                    offsets.put(partition, new OffsetAndMetadata(lastOffset + 1));
                }
                
                producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());
                
                // Commit transaction (atomic!)
                producer.commitTransaction();
                
            } catch (Exception e) {
                logger.error("Transaction failed", e);
                producer.abortTransaction();
            }
        }
        
        producer.close();
        consumer.close();
    }
}

Result: Exactly-once end-to-end (consume → transform → produce).

8. Common Anti-Patterns

Anti-Pattern 1: Creating Consumer Per Message

// ❌ BAD: Creates new consumer for each message
for (Message msg : messages) {
    KafkaConsumer consumer = new KafkaConsumer<>(props);
    consumer.poll(...);
    consumer.close();
}

Why bad: Huge overhead, defeats purpose of consumer groups.

Fix: Create one consumer, reuse.

Anti-Pattern 2: Blocking in Poll Loop

// ❌ BAD: Blocking call in poll loop
while (true) {
    ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord record : records) {
        slowDatabaseCall(record); // Takes 5 seconds!
    }
}
// Result: max.poll.interval.ms timeout!

Fix: Use separate processing thread or reduce batch size.

Anti-Pattern 3: Not Committing Offsets

// ❌ BAD: No commit
while (true) {
    ConsumerRecords records = consumer.poll(...);
    for (ConsumerRecord record : records) {
        processRecord(record);
    }
    // No commit! On restart, reprocess everything
}

Fix: Commit offsets regularly.

Anti-Pattern 4: Committing Before Processing

// ❌ BAD: At-most-once (data loss)
while (true) {
    ConsumerRecords records = consumer.poll(...);
    consumer.commitSync(); // Commit first!
    
    for (ConsumerRecord record : records) {
        processRecord(record); // Crash here = data lost
    }
}

Fix: Process then commit (at-least-once).

Anti-Pattern 5: Ignoring Rebalances

// ❌ BAD: No rebalance listener
consumer.subscribe(topics); // No listener!

// Can't commit on revoke
// Lose work on rebalance

Fix: Use ConsumerRebalanceListener to commit on revoke.

9. Production-Ready Consumer Template

Complete Implementation

public class ProductionConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final AtomicBoolean running = new AtomicBoolean(true);
    private final Metrics metrics;
    private final DeadLetterProducer dlq;
    
    public ProductionConsumer() {
        Properties props = new Properties();
        
        // Connection
        props.put("bootstrap.servers", "broker1:9092,broker2:9092,broker3:9092");
        props.put("group.id", "my-consumer-group");
        props.put("client.id", "consumer-" + InetAddress.getLocalHost().getHostName());
        
        // Serialization
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", ErrorHandlingDeserializer.class.getName());
        
        // Offsets
        props.put("enable.auto.commit", "false");
        props.put("auto.offset.reset", "earliest");
        
        // Performance (balanced)
        props.put("max.poll.records", 500);
        props.put("fetch.min.bytes", 1);
        props.put("max.partition.fetch.bytes", 1048576);
        
        // Timeouts
        props.put("session.timeout.ms", 10000);
        props.put("heartbeat.interval.ms", 3000);
        props.put("max.poll.interval.ms", 300000);
        
        // Modern features
        props.put("partition.assignment.strategy", 
            CooperativeStickyAssignor.class.getName());
        
        this.consumer = new KafkaConsumer<>(props);
        this.metrics = new Metrics();
        this.dlq = new DeadLetterProducer();
        
        // Shutdown hook
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            running.set(false);
            consumer.wakeup();
        }));
    }
    
    public void run() {
        try {
            consumer.subscribe(
                Collections.singletonList("orders"),
                new ConsumerRebalanceListener() {
                    @Override
                    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                        logger.info("Rebalance: revoking {}", partitions);
                        consumer.commitSync();
                        metrics.recordRebalance();
                    }
                    
                    @Override
                    public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                        logger.info("Rebalance: assigned {}", partitions);
                    }
                }
            );
            
            while (running.get()) {
                ConsumerRecords<String, String> records = 
                    consumer.poll(Duration.ofMillis(100));
                
                metrics.recordPoll(records.count());
                
                for (ConsumerRecord<String, String> record : records) {
                    try {
                        long startTime = System.currentTimeMillis();
                        
                        // Validate
                        if (record.value() == null) {
                            logger.warn("Null value at offset {}", record.offset());
                            dlq.send(record, new NullPointerException("Null value"));
                            continue;
                        }
                        
                        // Process
                        processRecord(record);
                        
                        // Metrics
                        long latency = System.currentTimeMillis() - startTime;
                        metrics.recordSuccess(latency);
                        
                    } catch (Exception e) {
                        logger.error("Processing failed", e);
                        metrics.recordFailure();
                        dlq.send(record, e);
                    }
                }
                
                // Commit
                consumer.commitAsync((offsets, exception) -> {
                    if (exception != null) {
                        logger.error("Commit failed", exception);
                        metrics.recordCommitFailure();
                    }
                });
            }
            
        } catch (WakeupException e) {
            if (running.get()) throw e;
        } finally {
            try {
                consumer.commitSync();
            } finally {
                consumer.close();
                dlq.close();
                logger.info("Consumer shut down cleanly");
            }
        }
    }
}

10. Summary Checklist

Consumer Pattern Decision Tree

What's your use case?

High Throughput:
└─ Use Pattern 1 (Batch Consumer)

Low Latency:
└─ Use Pattern 2 (Stream Processor)

Need Idempotence:
└─ Use Pattern 3 (Idempotent Consumer)

Need Parallelism + Ordering:
└─ Use Pattern 4 (Thread-Per-Partition)

Exactly-Once E2E:
└─ Use Pattern 7 (Transactional Consumer-Producer)

Summary

You learned:

• ✅ Batch consumer for high throughput
• ✅ Stream processor for low latency
• ✅ Idempotent consumer for duplicate prevention
• ✅ Thread-per-partition for parallel processing with ordering
• ✅ Graceful shutdown for safe termination
• ✅ Backpressure handling for downstream protection
• ✅ Exactly-once pipeline for transactional processing
• ✅ Common anti-patterns to avoid
• ✅ Production-ready template with best practices

Course Complete!

Congratulations! You've completed the Kafka Consumers course.

You now understand:

1. Consumer fundamentals and mental models
2. Consumer groups and coordination
3. Partition assignment strategies
4. Rebalancing mechanics
5. Offset management and commits
6. Delivery semantics
7. Poll loop internals
8. Performance tuning
9. Error handling and resilience
10. Monitoring and observability
11. Real-world patterns

Next steps:

• Apply these patterns to your projects
• Experiment with different configurations
• Monitor your consumers in production
• Contribute back to the community

Happy consuming! 🎉