7.1 Configuration & Performance Tuning

Module Overview

Goal: Teach safe, intentional consumer tuning based on understanding from previous modules.

What You'll Learn:

Fetch size configurations and their impact
max.poll.records tuning
Throughput vs latency trade-offs
Parallel processing strategies
Scaling consumer groups safely
Avoiding rebalance storms
Memory management

What's ASSUMED:

Offsets (Module 5)
Poll loop mechanics (Module 6.1)
Rebalancing (Module 4.1)

Key Insight: Tuning without understanding leads to problems. This module applies previous knowledge to performance optimization.

1. Fetch Configuration

fetch.min.bytes

props.put("fetch.min.bytes", 1024); // 1 KB (default: 1)

Purpose: Minimum data the broker should return per fetch request.

Behavior:

Broker waits until fetch.min.bytes available (up to fetch.max.wait.ms)
Reduces number of fetch requests
Increases throughput, adds latency

Tuning:

Low throughput, low latency:  fetch.min.bytes = 1 (default)
High throughput:              fetch.min.bytes = 1048576 (1 MB)

fetch.max.wait.ms

props.put("fetch.max.wait.ms", 500); // 500ms (default)

Purpose: Max time broker waits if fetch.min.bytes not reached.

Prevents: Infinite waiting when traffic is low.

fetch.max.bytes

props.put("fetch.max.bytes", 52428800); // 50 MB (default)

Purpose: Maximum data returned per fetch request (across all partitions).

Use when:

Need to limit memory usage
Network bandwidth constrained

max.partition.fetch.bytes

props.put("max.partition.fetch.bytes", 1048576); // 1 MB (default)

Purpose: Maximum data per partition per fetch.

Calculation:

Total fetch memory ≤ max.partition.fetch.bytes × partitions_assigned

Example:

max.partition.fetch.bytes = 1 MB
Consumer assigned 10 partitions
Max memory per fetch = 10 MB

2. max.poll.records

What It Controls

props.put("max.poll.records", 500); // Default

Purpose: Maximum records returned by single poll().

Tuning for Latency

// Low latency (fast poll loop)
props.put("max.poll.records", 50);

while (true) {
    records = consumer.poll(Duration.ofMillis(100)); // Returns max 50 records
    // Fast processing
    consumer.commitSync();
}

Result: Low latency, more frequent commits, lower throughput.

Tuning for Throughput

// High throughput (batch processing)
props.put("max.poll.records", 5000);

while (true) {
    records = consumer.poll(Duration.ofMillis(100)); // Returns up to 5000 records
    // Batch process
    consumer.commitSync();
}

Result: High throughput, higher latency, less frequent commits.

The Trade-off

max.poll.records	Poll Frequency	Latency	Throughput	Commit Overhead
50	High	Low	Lower	High
500 (default)	Medium	Medium	Good	Medium
5000	Low	High	Higher	Low

3. Timeout Configuration Matrix

The Configuration Relationship

heartbeat.interval.ms = 3000         (3 seconds)
session.timeout.ms = 10000           (10 seconds, > 3 × heartbeat)
max.poll.interval.ms = 300000        (5 minutes, >> session timeout)

Rule:

heartbeat.interval.ms < session.timeout.ms / 3 < max.poll.interval.ms

Configuration Profiles

Real-Time Profile (Fast Failover)

props.put("heartbeat.interval.ms", 1000);          // 1 second
props.put("session.timeout.ms", 6000);             // 6 seconds
props.put("max.poll.interval.ms", 60000);          // 1 minute
props.put("max.poll.records", 100);                // Small batches

Use when:

Fast failure detection critical
Low latency required
Fast processing guaranteed

High-Throughput Profile

props.put("heartbeat.interval.ms", 3000);          // 3 seconds
props.put("session.timeout.ms", 30000);            // 30 seconds
props.put("max.poll.interval.ms", 600000);         // 10 minutes
props.put("max.poll.records", 2000);               // Large batches

Use when:

Throughput more important than latency
Slower processing acceptable
Rare rebalances

Balanced Profile (Recommended)

props.put("heartbeat.interval.ms", 3000);          // 3 seconds (default)
props.put("session.timeout.ms", 10000);            // 10 seconds (default)
props.put("max.poll.interval.ms", 300000);         // 5 minutes (default)
props.put("max.poll.records", 500);                // Default

Use when:

Starting point for tuning
Balanced requirements
Standard use cases

4. Parallel Processing Strategies

Strategy 1: Single Consumer, Sequential Processing

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record); // Sequential
    }
}

Pros: Simple, maintains order
Cons: Slow, underutilizes CPU

Strategy 2: Single Consumer, Thread Pool

ExecutorService executor = Executors.newFixedThreadPool(10);

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    List<Future<?>> futures = new ArrayList<>();
    for (ConsumerRecord<String, String> record : records) {
        futures.add(executor.submit(() -> processRecord(record)));
    }
    
    // Wait for all to complete
    for (Future<?> future : futures) {
        future.get();
    }
    
    consumer.commitSync(); // After all processed
}

Pros: Parallel processing, better CPU utilization
Cons: Loses ordering within partition

Strategy 3: Multiple Consumers (Recommended)

// Run multiple consumer instances
for (int i = 0; i < numConsumers; i++) {
    new Thread(() -> {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("orders"));
        
        while (running) {
            ConsumerRecords<String, String> records = consumer.poll(...);
            for (ConsumerRecord<String, String> record : records) {
                processRecord(record);
            }
            consumer.commitSync();
        }
        
        consumer.close();
    }).start();
}

Pros:

Natural Kafka scaling
Ordering preserved per partition
Fault isolation

Cons:

More consumers needed
Limited by partition count

5. Scaling Consumer Groups

Optimal Consumer Count

Partitions: P
Consumers: C

Optimal: C = P (1:1 mapping)
Acceptable: C < P
Wasteful: C > P (idle consumers)

Example:

Topic: 12 partitions

1 consumer:  1 consumer × 12 partitions = Overloaded
3 consumers: 3 consumers × 4 partitions each = Good
6 consumers: 6 consumers × 2 partitions each = Better
12 consumers: 12 consumers × 1 partition each = Optimal
15 consumers: 12 active + 3 idle = Wasteful

Scaling Guidelines

Scale Up When:

Consumer lag increasing
Processing slow
CPU underutilized
C < P (room to add consumers)

Don't Scale When:

C = P already
Network is bottleneck
Processing is I/O bound
Need more partitions instead

Adding Partitions vs Adding Consumers

Scenario: 10 consumers, 10 partitions, still slow

Option A: Add more consumers
Result: 11th consumer sits idle (wasteful)

Option B: Add partitions (requires admin)
kafka-topics.sh --alter --topic orders --partitions 20
Result: Can now have 20 consumers (effective)

Key: Partitions set the upper limit for parallelism.

6. Memory Management

Consumer Memory Usage

Total memory = 
    fetch.max.bytes (fetch buffer)
  + max.partition.fetch.bytes × assigned_partitions (partition buffers)
  + Application processing memory
  + JVM overhead

Memory Tuning Example

// Scenario: 20 partitions assigned, 2 GB heap

// Option 1: Conservative
props.put("max.partition.fetch.bytes", 1048576);    // 1 MB
// Max fetch = 20 MB (20 partitions × 1 MB)

// Option 2: Aggressive
props.put("max.partition.fetch.bytes", 10485760);   // 10 MB
// Max fetch = 200 MB (20 partitions × 10 MB)

// Option 3: Balanced
props.put("max.partition.fetch.bytes", 5242880);    // 5 MB
// Max fetch = 100 MB (20 partitions × 5 MB)

7. Avoiding Rebalance Storms

Common Causes

Slow Processing

// Bad: Processing exceeds max.poll.interval.ms
for (record : records) { // 1000 records
    slowDatabaseCall(record); // 10 seconds per record!
}
// Total: 10,000 seconds → Timeout!

Fix: Reduce max.poll.records or increase max.poll.interval.ms.

GC Pauses

Long GC pause → Missed heartbeats → Rebalance

Fix:

Tune JVM GC settings
Increase session.timeout.ms
Reduce heap usage

Network Issues

Network partition → Lost heartbeats → Rebalance

Fix:

Increase session.timeout.ms
Investigate network stability
Use static membership

Rebalance Storm Prevention

// Configuration to prevent false positives
props.put("session.timeout.ms", 30000);        // 30s (vs default 10s)
props.put("heartbeat.interval.ms", 10000);     // 10s (vs default 3s)
props.put("max.poll.interval.ms", 600000);     // 10m (vs default 5m)
props.put("max.poll.records", 200);            // Reduce batch size

// Use static membership for controlled deployments
props.put("group.instance.id", "consumer-" + instanceId);

// Use cooperative rebalancing
props.put("partition.assignment.strategy", 
    "org.apache.kafka.clients.consumer.CooperativeStickyAssignor");

8. Performance Benchmarking

Metrics to Measure

// Throughput
long recordsProcessed = 0;
long startTime = System.currentTimeMillis();

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(...);
    recordsProcessed += records.count();
    
    // Every 10 seconds
    if (System.currentTimeMillis() - startTime > 10000) {
        double throughput = recordsProcessed / 10.0;
        System.out.println("Throughput: " + throughput + " records/sec");
        recordsProcessed = 0;
        startTime = System.currentTimeMillis();
    }
}

Latency Measurement

for (ConsumerRecord<String, String> record : records) {
    long produceTime = record.timestamp();
    long consumeTime = System.currentTimeMillis();
    long latency = consumeTime - produceTime;
    
    latencyHistogram.record(latency);
}

// Report p50, p95, p99
System.out.println("Latency p50: " + latencyHistogram.p50());
System.out.println("Latency p99: " + latencyHistogram.p99());

9. Tuning Decision Tree

What's your priority?

Throughput:
├─ Increase max.poll.records (2000+)
├─ Increase fetch.min.bytes (1 MB)
├─ Use batch processing
└─ Add more consumers (up to partition count)

Latency:
├─ Decrease max.poll.records (50-100)
├─ Decrease fetch.min.bytes (1 byte)
├─ Process records immediately
└─ Optimize processing code

Stability:
├─ Increase session.timeout.ms (30s)
├─ Increase max.poll.interval.ms (10m)
├─ Use static membership
└─ Use cooperative rebalancing

10. Configuration Checklist

Production-Ready Configuration

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-" + hostname);

// Serialization
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());

// Offset management
props.put("enable.auto.commit", "false"); // Manual commit
props.put("auto.offset.reset", "earliest"); // Or "latest"

// Performance (balanced profile)
props.put("max.poll.records", 500);
props.put("fetch.min.bytes", 1);
props.put("fetch.max.wait.ms", 500);
props.put("max.partition.fetch.bytes", 1048576); // 1 MB

// Timeouts (stable)
props.put("session.timeout.ms", 10000);          // 10s
props.put("heartbeat.interval.ms", 3000);        // 3s
props.put("max.poll.interval.ms", 300000);       // 5min

// Rebalancing (modern)
props.put("partition.assignment.strategy",
    CooperativeStickyAssignor.class.getName());

// Optional: Static membership for stateful consumers
// props.put("group.instance.id", "consumer-" + instanceId);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

Summary

You learned:

✅ Fetch configs control batch size and latency
✅ max.poll.records is throughput vs latency trade-off
✅ Timeout configs must follow the hierarchy
✅ Multiple consumers better than thread pools
✅ Scale consumers up to partition count
✅ Prevent rebalance storms with tuning
✅ Measure throughput and latency
✅ Use balanced profile as starting point

Next up: Error Handling & Resilience (Module 8.1) — Learn how to handle failures gracefully.