Overview
Kafka producers and consumers are deceptively simple to use but have many configuration options that directly affect throughput, latency, and delivery semantics. The defaults favor availability over durability — production systems almost always require explicit configuration for acks, retries, idempotence, and commit strategies.
Architecture
PRODUCER SEND PIPELINE
════════════════════════════════════════════════════════
producer.send(record)
│
Serializer (key + value)
│
Partitioner → determines which partition
│ Default: hash(key) % numPartitions
│ Custom: implement Partitioner interface
│
RecordAccumulator (per-partition buffer)
│ Batches records for linger.ms or until batch.size full
│
Sender Thread (background)
│ Drains batches → NetworkClient → Broker
│
Broker ACK → callback(metadata, exception)
THROUGHPUT KNOBS:
batch.size = 16384 (16KB) → increase for higher throughput
linger.ms = 0 → increase (5-10ms) to fill batches
buffer.memory = 32MB → total producer buffer; block if full
compression.type = none → lz4 / snappy reduces network I/O
CONSUMER GROUP MECHANICS
════════════════════════════════════════════════════════
Topic: orders (12 partitions)
Consumer Group "order-processor" (3 consumers):
┌────────────┬────────────────────────────────┐
│ Consumer 1 │ P0, P1, P2, P3 │
│ Consumer 2 │ P4, P5, P6, P7 │
│ Consumer 3 │ P8, P9, P10, P11 │
└────────────┴────────────────────────────────┘
Add Consumer 4: rebalance → each consumer gets 3 partitions
Remove Consumer 2: rebalance → others take its partitions
KEY: # consumers ≤ # partitions (extra consumers idle)
Scale up: increase partitions FIRST, then add consumers
REBALANCING STRATEGIES
════════════════════════════════════════════════════════
Eager (STOP-THE-WORLD, default pre-Kafka 2.4):
All consumers stop, revoke ALL partitions, re-assign from scratch
Downtime during rebalance (can be seconds for large groups)
Cooperative/Incremental (Kafka 2.4+):
Only re-assign the partitions that need to move
Other partitions keep processing during rebalance
Enable: partition.assignment.strategy = CooperativeStickyAssignor
Technical Implementation
Spring Kafka Consumer with Manual Commit
@Component
public class OrderConsumer {
@KafkaListener(
topics = "orders",
groupId = "order-processor",
containerFactory = "kafkaListenerContainerFactory",
concurrency = "3" // 3 threads, each consuming some partitions
)
public void consume(
List<ConsumerRecord<String, OrderEvent>> records,
Acknowledgment ack // manual commit
) {
try {
// Process all records in the batch
records.forEach(record -> {
log.debug("Processing order {} from partition {} offset {}",
record.key(), record.partition(), record.offset());
orderService.process(record.value());
});
// Commit AFTER all records processed
ack.acknowledge();
} catch (Exception ex) {
log.error("Batch processing failed, will retry: {}", ex.getMessage());
// Don't acknowledge → offsets not committed → messages retried
// Use DeadLetterPublishingRecoverer for poison pills
}
}
}
// Factory configuration
@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, OrderEvent> factory =
new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
factory.setBatchListener(true); // batch processing
factory.getContainerProperties().setPollTimeout(3000);
// Dead letter topic for messages that fail N times
factory.setCommonErrorHandler(
new DefaultErrorHandler(
new DeadLetterPublishingRecoverer(kafkaTemplate(),
(record, ex) -> new TopicPartition("orders.DLT", record.partition())),
new FixedBackOff(1000L, 3L) // 3 retries, 1s apart
)
);
return factory;
}
Idempotent Producer with Transactions
// Transactional producer: read-process-write exactly-once
@Bean
public KafkaTransactionManager<String, OrderEvent> kafkaTransactionManager(
ProducerFactory<String, OrderEvent> pf) {
return new KafkaTransactionManager<>(pf);
}
@Service
@Transactional("kafkaTransactionManager")
public class OrderProcessor {
private final KafkaTemplate<String, OrderEvent> template;
// Reads from input topic, transforms, writes to output topic — exactly once
@KafkaListener(topics = "orders-raw")
public void process(ConsumerRecord<String, RawOrderEvent> record,
@Header(KafkaHeaders.ACKNOWLEDGMENT) Acknowledgment ack) {
OrderEvent enriched = enrich(record.value());
// Sent within the same transaction as the offset commit
template.send("orders-enriched", record.key(), enriched);
// @Transactional handles offset commit + produce atomically
// If send fails → transaction rolls back → offset not committed → retry
}
}
Partition Assignment and Rebalance Listener
@Component
public class RebalanceAwareConsumer {
@KafkaListener(topics = "events", groupId = "processor")
public void listen(ConsumerRecord<String, Event> record) {
processEvent(record.value());
}
// Hook into partition assignment lifecycle
@Bean
public ConsumerRebalanceListener rebalanceListener() {
return new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
log.info("Partitions revoked: {} — flushing in-progress work", partitions);
// Flush any in-memory buffers for revoked partitions
// Commit any pending offsets before partitions move to another consumer
flushAndCommit(partitions);
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
log.info("Partitions assigned: {} — initializing state", partitions);
// Load initial state for new partitions (e.g., read from changelog topic)
initializeState(partitions);
}
};
}
}
Interview Preparation
Q: What is the difference between at-least-once and exactly-once delivery in Kafka?
At-least-once: consumer processes a message, then commits the offset. If the consumer crashes after processing but before committing, it restarts from the last committed offset and processes the same message again. The message is processed at least once, possibly more. Your consumer must be idempotent (same message processed twice has the same effect as once). Exactly-once (Kafka's EXACTLY_ONCE_V2): uses transactions — the producer assigns a sequence number to each message, the broker deduplicates, and offset commits are atomic with message production. Much more complex and slower. Use at-least-once + idempotent consumers for most cases; exactly-once for financial systems where duplicate processing is genuinely unacceptable.
Q: What causes a consumer group rebalance and how do you minimize disruption?
Rebalances trigger on: consumer joining (new instance), consumer leaving (crash, graceful shutdown), partition count change, session timeout (session.timeout.ms). Disruptions: during a rebalance, all consumers pause. Minimize by: (1) Use CooperativeStickyAssignor — only moves the partitions that need to change, others keep processing. (2) Increase session.timeout.ms to avoid spurious timeouts on GC pauses or slow processing. (3) Set max.poll.interval.ms to match your worst-case processing time — if poll() isn't called within this interval, the consumer is assumed dead and a rebalance triggers. (4) Use static group membership (group.instance.id) — a consumer that disconnects and reconnects within session.timeout.ms won't trigger a rebalance.
Q: How many consumers can you have in a consumer group?
A maximum of one consumer per partition can actively receive messages. With 12 partitions and 12 consumers, each consumer gets exactly one partition. A 13th consumer would be idle (no partition to assign to). To scale consumption, increase partitions (pre-plan — you can't decrease them). To handle processing bottlenecks, you can process in parallel within a consumer (multiple worker threads per consumer instance), or scale out consumers up to the partition count.