5.2 Commit Strategies

Module Overview

Goal: Explain how and when offsets are committed — the mechanics of saving consumption progress.

What You'll Learn:

• Auto-commit vs manual commit
• Synchronous vs asynchronous commit
• Commit timing trade-offs
• Commit on process vs commit on poll
• Commit strategies for different scenarios
• Performance implications of commits

What's NOT Covered Here:

• What offsets are (covered in Module 5.1)
• Delivery semantics implications (covered in Module 5.3)
• Error recovery strategies (covered in Module 8.1)

Prerequisite: Module 5.1 (Offsets Explained) — You must understand current position vs committed offset.

Key Insight: Commit strategy determines the trade-off between performance and safety.

1. Why Committing Matters

The Problem

Consumer reads offsets 0-9:
[0][1][2][3][4][5][6][7][8][9]
            ↑
       Processing...

Consumer crashes before committing!

Consumer restarts:
[0][1][2][3][4][5][6][7][8][9]
 ↑
 Starts from beginning (last committed: nothing)

Result: Messages 0-9 processed twice!

Without committing:

• Progress not saved
• Restart means reprocessing from last committed offset (or auto.offset.reset)
• Risk of duplicate processing

What Committing Does

Consumer reads offsets 0-9:
[0][1][2][3][4][5][6][7][8][9]

Process offsets 0-4
Commit offset 5  ← Save progress!

Consumer crashes

Consumer restarts:
[0][1][2][3][4][5][6][7][8][9]
                ↑
            Resume from 5

Result: Messages 0-4 not reprocessed

Committing saves:

• Consumption progress
• Resume point for restarts
• Recovery position for failures

2. Auto-Commit (Default)

How Auto-Commit Works

// Default configuration
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "5000"); // 5 seconds

Behavior:

• Consumer automatically commits offsets in the background
• Commits happen every auto.commit.interval.ms
• Commits during poll() calls
• No explicit commit needed

Auto-Commit Timeline

Time  Consumer Action              Committed Offset
────┼────────────────────────────┼─────────────────
  0s│ poll() returns offsets 0-9 │ 0
  2s│ Processing...              │ 0
  5s│ poll() returns 10-19       │ 10 (auto-commit!)
  7s│ Processing...              │ 10
 10s│ poll() returns 20-29       │ 20 (auto-commit!)

Key: Commit happens during poll(), not immediately after processing.

Auto-Commit Example

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("enable.auto.commit", "true");  // Auto-commit enabled
props.put("auto.commit.interval.ms", "5000"); // Every 5 seconds

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("orders"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    // ↑ Auto-commit may happen here!
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record); // Your logic
    }
    // No explicit commit needed
}

Auto-Commit Pros and Cons

Pros:

• ✅ Simple — No commit code needed
• ✅ Automatic — Can't forget to commit
• ✅ Low overhead — Batches commits

Cons:

• ❌ No control over timing — Commits at fixed intervals
• ❌ Duplicate risk — Crash between commit and processing completion
• ❌ At-most-once risk — Commit before processing completes

Use when:

• Simplicity preferred
• Duplicate processing acceptable
• Idempotent processing

Auto-Commit Pitfall

Time  Action                     Committed    Risk
────┼──────────────────────────┼───────────┼────────────────────
  0s│ poll() → offsets 0-99    │ 0         │
  1s│ Processing 0-99...       │ 0         │
  5s│ poll() → offsets 100-199 │ 100       │ ← Commit happens!
  6s│ Still processing 0-99... │ 100       │ ← But not done!
  7s│ CRASH!                   │ 100       │

Restart:
  0s│ Resume from offset 100   │           │ ← Lost 0-99!

Problem: Auto-commit happens during poll(), which may be before processing completes.

Result: At-most-once delivery (messages can be lost).

3. Manual Commit

Disabling Auto-Commit

props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

Now you control:

• When commits happen
• What offsets to commit
• Error handling on commit failure

Synchronous Commit

props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    
    // Commit after processing
    consumer.commitSync();  // ← Blocks until commit succeeds
}

commitSync() behavior:

• Blocks until broker acknowledges
• Retries on retriable errors
• Throws exception on non-retriable errors
• Guarantees commit before returning

Synchronous Commit Timeline

Time  Consumer Action           Broker
────┼─────────────────────────┼───────────────────
  0s│ poll() → offsets 0-9    │
  1s│ Process 0-9             │
  2s│ commitSync() ──────────→│
  3s│ (waiting...)            │ Write offset
  4s│ ←────── ACK ────────────│
  5s│ Continue                │

Key: Consumer waits for broker acknowledgment.

Synchronous Commit Pros and Cons

Pros:

• ✅ Safe — Commit confirmed before proceeding
• ✅ Retry logic — Automatic retries on failure
• ✅ Predictable — Know exactly when commit happens

Cons:

• ❌ Blocking — Slows down consumption
• ❌ Higher latency — Wait for broker on every commit
• ❌ Lower throughput — Can't do other work during commit

Use when:

• Safety more important than performance
• Can tolerate higher latency
• Simple error handling preferred

Asynchronous Commit

props.put("enable.auto.commit", "false");

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    
    // Commit asynchronously (non-blocking)
    consumer.commitAsync();  // ← Returns immediately
}

commitAsync() behavior:

• Non-blocking — Returns immediately
• No retries — Fire and forget
• Optional callback — Get notified on completion
• Higher throughput — No waiting

Asynchronous Commit with Callback

consumer.commitAsync((offsets, exception) -> {
    if (exception != null) {
        logger.error("Commit failed for offsets {}", offsets, exception);
    } else {
        logger.info("Commit succeeded for offsets {}", offsets);
    }
});

Asynchronous Commit Pros and Cons

Pros:

• ✅ Non-blocking — Better throughput
• ✅ Low latency — No waiting for broker
• ✅ Efficient — Can process while committing

Cons:

• ❌ No retry — Failed commits not retried
• ❌ Uncertainty — Don't know if commit succeeded
• ❌ Out-of-order risk — Later commits may succeed while earlier fail

Use when:

• Throughput critical
• Can tolerate occasional commit failures
• Monitoring commit success rate

4. Commit Timing Strategies

Strategy 1: Commit After Every Batch

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

Characteristics:

• Small commit window
• More frequent commits
• Lower duplicate risk
• Higher overhead

Use when: Processing fast, batches small.

Strategy 2: Commit After Every Message

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
        
        // Commit after each message
        consumer.commitSync(Collections.singletonMap(
            new TopicPartition(record.topic(), record.partition()),
            new OffsetAndMetadata(record.offset() + 1)
        ));
    }
}

Characteristics:

• Slowest — Commit per message
• Minimal duplicate risk
• Very high overhead
• Only for critical data

Use when: Absolutely cannot tolerate duplicates, idempotence not possible.

Strategy 3: Commit Periodically (Time-Based)

long lastCommit = System.currentTimeMillis();
int commitInterval = 5000; // 5 seconds

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    
    if (System.currentTimeMillis() - lastCommit > commitInterval) {
        consumer.commitSync();
        lastCommit = System.currentTimeMillis();
    }
}

Characteristics:

• Predictable commit frequency
• Balance between safety and performance
• Independent of message rate

Strategy 4: Commit After N Messages

int messagesProcessed = 0;
int commitThreshold = 1000;

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
        messagesProcessed++;
        
        if (messagesProcessed >= commitThreshold) {
            consumer.commitSync();
            messagesProcessed = 0;
        }
    }
}

Characteristics:

• Adapts to message rate
• Fixed duplicate window (N messages)
• Variable commit frequency

Strategy 5: Mixed Async + Final Sync

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    
    consumer.commitAsync(); // Fast, non-blocking
}

// On shutdown
consumer.commitSync(); // Ensure last commit succeeds

Best of both worlds:

• Async during normal operation (performance)
• Sync on shutdown (safety)

5. Commit Strategies Comparison

6. Commit on Rebalance

The Problem

Consumer assigned partitions 0, 1, 2
Processing partition 0 offset 100
Rebalance triggered!

Consumer loses partition 0
Last committed offset: 50

New consumer gets partition 0
Starts from offset 50

Result: Offsets 50-100 processed twice!

Solution: Commit before losing partitions.

Using ConsumerRebalanceListener

consumer.subscribe(
    Collections.singletonList("orders"),
    new ConsumerRebalanceListener() {
        @Override
        public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // Commit before losing partitions
            consumer.commitSync();
            logger.info("Committed offsets before revocation: {}", partitions);
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            logger.info("Partitions assigned: {}", partitions);
        }
    }
);

Best practice: Always commit in onPartitionsRevoked().

7. Commit-Specific Offsets

Commit Last Processed Offset

for (ConsumerRecord<String, String> record : records) {
    processRecord(record);
    
    // Commit this specific partition/offset
    Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
    offsets.put(
        new TopicPartition(record.topic(), record.partition()),
        new OffsetAndMetadata(record.offset() + 1) // +1 because commit is "next to read"
    );
    
    consumer.commitSync(offsets);
}

Note: Committed offset is record.offset() + 1 because it represents the next offset to read.

8. Handling Commit Failures

Synchronous Commit Error Handling

try {
    consumer.commitSync();
} catch (CommitFailedException e) {
    // Rebalance happened, lost partitions
    logger.error("Commit failed (rebalanced)", e);
} catch (Exception e) {
    // Other error (network, etc.)
    logger.error("Commit failed", e);
}

Asynchronous Commit Error Handling

consumer.commitAsync((offsets, exception) -> {
    if (exception instanceof RetriableException) {
        // Transient error - could retry
        logger.warn("Retriable commit error", exception);
    } else if (exception != null) {
        // Permanent error
        logger.error("Commit failed", exception);
    }
});

Note: commitAsync() does not automatically retry. You must implement retry logic if needed.

9. Performance Considerations

Commit Overhead

commitSync() overhead:
- Network round trip: ~5-50ms
- Broker write: ~1-5ms
- Total: ~10-100ms per commit

For 10,000 msgs/sec:
- Commit per message: 10,000 commits/sec → Impossible!
- Commit per batch (100 msgs): 100 commits/sec → Feasible
- Commit every 5s: 0.2 commits/sec → Very low overhead

Recommendation: Commit per batch or periodically, not per message.

Commit Frequency Trade-offs

High Frequency (every message):
✅ Low duplicate risk
❌ Very slow
❌ High network overhead

Medium Frequency (every batch):
✅ Balanced
✅ Acceptable duplicates
✅ Good throughput

Low Frequency (every 5-10s):
✅ Fast
✅ Low overhead
❌ Higher duplicate risk on failure

10. Recommended Strategies

For High-Throughput Systems

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    
    // Async commit (non-blocking)
    consumer.commitAsync();
}

// On shutdown
consumer.commitSync(); // Ensure final commit
consumer.close();

For High-Reliability Systems

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
    }
    
    // Sync commit (guaranteed)
    consumer.commitSync();
}

consumer.close();

For Balanced Systems (Recommended)

int commitCount = 0;
int commitThreshold = 100;

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, String> record : records) {
        processRecord(record);
        commitCount++;
        
        if (commitCount >= commitThreshold) {
            consumer.commitAsync();
            commitCount = 0;
        }
    }
}

// Final sync commit
consumer.commitSync();
consumer.close();

11. Mental Model Summary

Key Concepts

COMMIT = Save consumption progress

Types:
├─ Auto-commit: Automatic, periodic
├─ Manual sync: Blocking, safe
└─ Manual async: Non-blocking, fast

Timing:
├─ Per message (slow, safe)
├─ Per batch (balanced)
└─ Periodic (fast, less safe)

Strategy:
└─ Trade-off: Performance ↔ Safety

Commit Decision Tree

Do you need maximum throughput?
├─ YES → commitAsync()
└─ NO → Do you need guaranteed commits?
         ├─ YES → commitSync()
         └─ NO → Auto-commit

12. What's Next?

Now that you understand how to commit offsets, the next module will cover:

• Module 5.3: How commit strategy affects delivery semantics (at-most-once, at-least-once, exactly-once)

Summary

You learned:

• ✅ Auto-commit: Simple, automatic, less control
• ✅ commitSync(): Blocking, safe, slower
• ✅ commitAsync(): Non-blocking, fast, no retries
• ✅ Commit timing strategies (per message, per batch, periodic)
• ✅ Always commit in onPartitionsRevoked()
• ✅ Use mixed async+sync for best balance
• ✅ Commit frequency is a performance vs safety trade-off

Next up: Consumer Delivery Semantics (Module 5.3) — Learn how commits determine at-most-once, at-least-once, and exactly-once guarantees.