Apache Kafka Architecture: A Deep Dive into Distributed Event Streaming
Architecture Diagram: Kafka Cluster Topology
╔══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
║ ║
║ ███████████████████████████████████████████████████████████████ ║
║ █ KAFKA CLUSTER (kafka-cluster-01) █ ║
║ █ █ ║
║ █ ┌───────────────────────────────────────────────────────────────────────────────────────┐ ║
║ █ │ ZOOKEEPER ENSEMBLE │ ║
║ █ │ │ ║
║ █ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ ║
║ █ │ │ ZK Node 1 │◄──►│ ZK Node 2 │◄──►│ ZK Node 3 │ │ ║
║ █ │ │ (Leader) │ │ (Follower) │ │ (Follower) │ │ ║
║ █ │ │ :2181 │ │ :2181 │ │ :2181 │ │ ║
║ █ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ ║
║ █ │ │ │ │ │ ║
║ █ └──────────┼────────────────────┼────────────────────┼─────────────────────────────────┘ ║
║ █ │ │ │ ║
║ █ ▼ ▼ ▼ ║
║ █ ┌───────────────────────────────────────────────────────────────────────────────────────┐ ║
║ █ │ BROKER LAYER │ ║
║ █ │ │ ║
║ █ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ ║
║ █ │ │ Broker 0 │ │ Broker 1 │ │ Broker 2 │ │ Broker 3 │ │ ║
║ █ │ │ (Controller)│ │ (Standby) │ │ (Standby) │ │ (Standby) │ │ ║
║ █ │ │ :9092 │ │ :9092 │ │ :9092 │ │ :9092 │ │ ║
║ █ │ │ ┌──────┐ │ │ ┌──────┐ │ │ ┌──────┐ │ │ ┌──────┐ │ │ ║
║ █ │ │ │Topic │ │ │ │Topic │ │ │ │Topic │ │ │ │Topic │ │ │ ║
║ █ │ │ │ Part │ │ │ │ Part │ │ │ │ Part │ │ │ │ Part │ │ │ ║
║ █ │ │ │itions│ │ │ │itions│ │ │ │itions│ │ │ │itions│ │ │ ║
║ █ │ │ └──────┘ │ │ └──────┘ │ │ └──────┘ │ │ └──────┘ │ │ ║
║ █ │ │ ┌──────┐ │ │ ┌──────┐ │ │ ┌──────┐ │ │ ┌──────┐ │ │ ║
║ █ │ │ │Commit│ │ │ │Commit│ │ │ │Commit│ │ │ │Commit│ │ │ ║
║ █ │ │ │ Log │ │ │ │ Log │ │ │ │ Log │ │ │ │ Log │ │ │ ║
║ █ │ │ └──────┘ │ │ └──────┘ │ │ └──────┘ │ │ └──────┘ │ │ ║
║ █ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ ║
║ █ │ │ │ │ │ │ ║
║ █ └─────────┼────────────────┼────────────────┼────────────────┼──────────────────────────┘ ║
║ █ │ │ │ │ ║
║ █ ▼ ▼ ▼ ▼ ║
║ █ ┌───────────────────────────────────────────────────────────────────────────────────────┐ ║
║ █ │ STORAGE LAYER │ ║
║ █ │ │ ║
║ █ │ ┌─────────────────────────────────────────────────────────────────────────────┐ │ ║
║ █ │ │ /var/kafka/data/ │ │ ║
║ █ │ │ ├── __consumer_offsets/ │ │ ║
║ █ │ │ ├── orders-topic-0/ │ │ ║
║ █ │ │ │ ├── 00000000000000000000.log │ │ ║
║ █ │ │ │ ├── 00000000000000000000.index │ │ ║
║ █ │ │ │ └── 00000000000000000000.timeindex │ │ ║
║ █ │ │ ├── orders-topic-1/ │ │ ║
║ █ │ │ ├── orders-topic-2/ │ │ ║
║ █ │ │ ├── user-events-topic-0/ │ │ ║
║ █ │ │ └── user-events-topic-1/ │ │ ║
║ █ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ║
║ █ └───────────────────────────────────────────────────────────────────────────────────────┘ ║
║ █ ║
║ ████████████████████████████████████████████████████████████████████████████████████████████████████ ║
║ ║
╚══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝
Architecture Diagram: Partition Distribution & Replication
╔══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
║ ║
║ TOPIC: "order-events" (3 Partitions, Replication Factor = 3) ║
║ ║
║ ┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐ ║
║ │ │ ║
║ │ PARTITION 0 PARTITION 1 PARTITION 2 │ ║
║ │ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ ║
║ │ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │ ║
║ │ │ │ LEADER │ │ │ │ LEADER │ │ │ │ LEADER │ │ │ ║
║ │ │ │ Broker 0 │ │ │ │ Broker 1 │ │ │ │ Broker 2 │ │ │ ║
║ │ │ │ (Preferred) │ │ │ │ (Preferred) │ │ │ │ Preferred) │ │ │ ║
║ │ │ └───────┬───────┘ │ │ └───────┬───────┘ │ │ └───────┬───────┘ │ │ ║
║ │ │ │ │ │ │ │ │ │ │ │ ║
║ │ │ ┌────┴────┐ │ │ ┌────┴────┐ │ │ ┌────┴────┐ │ │ ║
║ │ │ ▼ ▼ │ │ ▼ ▼ │ │ ▼ ▼ │ │ ║
║ │ │ ┌──────┐ ┌──────┐ │ │ ┌──────┐ ┌──────┐ │ │ ┌──────┐ ┌──────┐ │ │ ║
║ │ │ │ISRs: │ │ISRs: │ │ │ │ISRs: │ │ISRs: │ │ │ │ISRs: │ │ISRs: │ │ │ ║
║ │ │ │ 0,1,2│ │ 0,1,2│ │ │ │ 0,1,2│ │ 0,1,2│ │ │ │ 0,1,2│ │ 0,1,2│ │ │ ║
║ │ │ └──────┘ └──────┘ │ │ └──────┘ └──────┘ │ │ └──────┘ └──────┘ │ │ ║
║ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ║
║ │ │ ▼ ▼ │ │ ▼ ▼ │ │ ▼ ▼ │ │ ║
║ │ │ ┌──────┐ ┌──────┐ │ │ ┌──────┐ ┌──────┐ │ │ ┌──────┐ ┌──────┐ │ │ ║
║ │ │ │Rep 1 │ │Rep 2 │ │ │ │Rep 0 │ │Rep 2 │ │ │ │Rep 0 │ │Rep 1 │ │ │ ║
║ │ │ │Brkr 1│ │Brkr 2│ │ │ │Brkr 0│ │Brkr 2│ │ │ │Brkr 0│ │Brkr 1│ │ │ ║
║ │ │ └──────┘ └──────┘ │ │ └──────┘ └──────┘ │ │ └──────┘ └──────┘ │ │ ║
║ │ └──────────────────┘ └──────────────────┘ └──────────────────┘ │ ║
║ │ │ ║
║ │ Offset: 0 ──► 1000 Offset: 0 ──► 850 Offset: 0 ──► 1200 │ ║
║ │ (Last: 1000) (Last: 850) (Last: 1200) │ ║
║ │ │ ║
║ └─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘ ║
║ ║
╚══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝
Architecture Diagram: Consumer Group Coordination
╔══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╗
║ ║
║ CONSUMER GROUP: "order-processing-service" (Group ID: order-service-01) ║
║ ║
║ ┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐ ║
║ │ │ ║
║ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ ║
║ │ │ Consumer 1 │ │ Consumer 2 │ │ Consumer 3 │ │ ║
║ │ │ ┌───────────┐ │ │ ┌───────────┐ │ │ ┌───────────┐ │ │ ║
║ │ │ │ Thread 1 │ │ │ │ Thread 1 │ │ │ │ Thread 1 │ │ │ ║
║ │ │ │ Thread 2 │ │ │ │ Thread 2 │ │ │ │ Thread 2 │ │ │ ║
║ │ │ │ Thread 3 │ │ │ │ Thread 3 │ │ │ │ Thread 3 │ │ │ ║
║ │ │ └───────────┘ │ │ └───────────┘ │ │ └───────────┘ │ │ ║
║ │ │ Client ID: │ │ Client ID: │ │ Client ID: │ │ ║
║ │ │ consumer-1 │ │ consumer-2 │ │ consumer-3 │ │ ║
║ │ └────────┬────────┘ └────────┬────────┘ └────────┬────────┘ │ ║
║ │ │ │ │ │ ║
║ │ │ │ │ │ ║
║ │ ▼ ▼ ▼ │ ║
║ │ ┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────┐ │ ║
║ │ │ GROUP COORDINATOR (Broker 0) │ │ ║
║ │ │ ┌─────────────────────────────────────────────────────────────────────────────────────────────────────┐ │ │ ║
║ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ ║
║ │ │ │ │ Heartbeat │ │ Rebalance │ │ Offset │ │ Membership │ │ Assignment │ │ │ │ ║
║ │ │ │ │ Handler │ │ Protocol │ │ Manager │ │ Tracker │ │ Strategy │ │ │ │ ║
║ │ │ │ └──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘ └──────────────┘ │ │ │ ║
║ │ │ └─────────────────────────────────────────────────────────────────────────────────────────────────────┘ │ │ ║
║ │ └─────────────────────────────────────────────────────────────────────────────────────────────────────────────┘ │ ║
║ │ │ ║
║ │ PARTITION ASSIGNMENTS: │ ║
║ │ ┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────┐ │ ║
║ │ │ │ │ ║
║ │ │ Consumer 1 ◄── Partition 0 (Leader: Broker 0) │ │ ║
║ │ │ Consumer 2 ◄── Partition 1 (Leader: Broker 1) │ │ ║
║ │ │ Consumer 3 ◄── Partition 2 (Leader: Broker 2) │ │ ║
║ │ │ │ │ ║
║ │ └─────────────────────────────────────────────────────────────────────────────────────────────────────────────┘ │ ║
║ │ │ ║
║ └─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘ ║
║ ║
╚══════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════════╝
Formal Definitions
DfPartition
A partition is an ordered, immutable sequence of records that is continually appended to. Each record in a partition is assigned a sequential identifier called an offset. Partitions are the unit of parallelism in Kafka — the number of partitions determines the maximum consumer parallelism within a consumer group. Formally, a topic T is decomposed as T = {P_0, P_1, ..., P_{n-1}} where n is the partition count, and each P_i is a totally ordered log.
DfReplica
A replica is a fault-tolerant copy of a partition stored on a different broker. Each partition with replication factor R maintains R replicas. One replica is elected as the leader (handles all reads/writes), while the remaining R-1 are followers that replicate from the leader. Replicas ensure durability: if a broker fails, another replica can be promoted to leader with zero data loss.
DfIn-Sync Replicas (ISR)
The In-Sync Replica set (ISR) is the subset of replicas that are fully caught up with the leader. A replica is in the ISR if it has fetched the most recent messages within replica.lag.time.max.ms. The ISR is dynamic: replicas that fall behind are removed, and replicas that catch up are re-added. Writes are only acknowledged to the producer when all ISR replicas have persisted the record (when acks=all and min.insync.replicas is satisfied).
DfConsumer Group
A consumer group is a set of consumers that jointly consume messages from a topic. Each partition is consumed by exactly one consumer within the group, enabling load-balanced parallel consumption. The Group Coordinator broker manages membership and triggers rebalancing when consumers join or leave. Formally, for a topic with partitions P = {P_0, ..., P_{n-1}} and a group with consumers C = {C_0, ..., C_{m-1}}, the assignment function f: P → C maps each partition to exactly one consumer, where m ≤ n for full utilization.
Detailed Explanation
Apache Kafka is a distributed event streaming platform capable of handling trillions of events per day with sub-millisecond latency. At its core, Kafka operates as a distributed commit log, where data is persisted in an append-only fashion across a cluster of brokers. The fundamental architectural principle is decoupling of producers and consumers through a publish-subscribe model that enables horizontal scalability while maintaining strict ordering guarantees within partitions.
The Broker Layer consists of a cluster of servers (brokers) that each maintain partition data and serve client requests. Each broker has a unique identifier (broker.id) and listens on a configured port (default: 9092). The brokers communicate with each other for replication and with ZooKeeper (or KRaft controller) for cluster metadata management. When a producer sends a message, the broker acknowledges receipt based on the configured acknowledgment policy (acks=0, 1, or all). The broker stores the message in its local commit log, which is a sequence of files on disk optimized for sequential I/O operations. Each partition is a separate directory containing multiple log segment files, with only the active segment accepting writes.
Topics serve as logical categories or feed names for messages. Each topic is divided into multiple partitions, which are the unit of parallelism in Kafka. The number of partitions determines the maximum number of consumer instances that can read from the topic in parallel within a consumer group. Partitions are distributed across brokers using a consistent hashing algorithm, ensuring even distribution of load. Each partition maintains its own commit log, and messages within a partition are assigned monotonically increasing offsets. This design enables Kafka to achieve extremely high throughput by parallelizing both writes (across partitions) and reads (via consumer groups).
Replication provides fault tolerance and high availability. Each partition has a configurable replication factor (typically 3 for production), meaning data is replicated across multiple brokers. One broker is elected as the partition leader, handling all reads and writes for that partition. Follower replicas continuously pull data from the leader to stay synchronized. The set of in-sync replicas (ISR) includes all replicas that are fully caught up with the leader. If a follower falls too far behind (controlled by replica.lag.time.max.ms), it is removed from the ISR. When the leader fails, a new leader is elected from the ISR, ensuring data consistency. The ISR mechanism balances durability guarantees with availability, as the leader can continue serving requests as long as the minimum ISR size is maintained.
Consumer Groups enable load-balanced consumption across multiple consumers. Each consumer in a group is assigned a subset of partitions, ensuring that each partition is consumed by exactly one consumer within the group. The Group Coordinator (a broker) manages the group membership and triggers rebalancing when consumers join or leave. The rebalancing process uses protocols like RangeAssignor, RoundRobinAssignor, or CooperativeStickyAssignor to distribute partitions. Consumer offsets are tracked in the __consumer_offsets topic, allowing consumers to resume from their last committed position. This design enables horizontal scaling of consumption while maintaining ordering guarantees within partitions.
The Storage Layer uses a highly optimized append-only log structure. Each log segment consists of three files: .log (message data), .index (offset index), and .timeindex (timestamp index). The index files enable efficient random access to specific offsets without scanning the entire log. Kafka uses a page cache aggressively, leveraging the operating system's memory management for read performance. The log compaction feature allows Kafka to retain only the latest value for each key, enabling topics to serve as a materialized view of the latest state. Retention policies can be time-based, size-based, or compacted, providing flexibility in data lifecycle management.
Key Formulas
Replication Lag Model
Here,
- =Replication lag at time t (bytes)
- =Data written to leader at time t
- =Data fetched by follower at time t
Consumer Lag
Here,
- =Latest offset in the partition log
- =Last committed consumer offset
Partition count directly determines maximum parallelism. For a topic with N partitions, at most N consumers in a group can process in parallel. Beyond N consumers, additional consumers remain idle. Plan partition count based on target throughput: N_partitions ≥ T_target / T_broker.
ThCAP Theorem Applied to Kafka
Kafka's partition-level design offers a configurable trade-off between consistency and availability:
- With
acks=all,min.insync.replicas=2,unclean.leader.election.enable=false: Kafka prioritizes consistency over availability. If fewer thanmin.insync.replicasare available, writes are rejected (CP behavior). - With
acks=1orunclean.leader.election.enable=true: Kafka prioritizes availability over consistency. A partition can continue serving even if it means potential data loss (AP behavior). In practice, most production deployments choose CP (consistency-first) to prevent data loss.
ThOrdering Guarantee Theorem
Kafka guarantees total ordering within a partition but only partial ordering across partitions. If a producer sends messages m_1, m_2, m_3 to the same partition, a consumer will always observe them in that order. However, for messages across different partitions, no ordering guarantee exists. To achieve ordering on a key k, all messages with key k must be routed to the same partition via hash(k) mod N_partitions.
Key Concepts Table
| Component | Description | Default Value | Production Recommendation |
|---|---|---|---|
| Broker | Individual Kafka server node | Port 9092 | 3+ nodes per cluster |
| Topic | Logical message category | N/A | Use kebab-case naming |
| Partition | Unit of parallelism within topic | 1 | 6-12 per topic (based on throughput) |
| Replica | Copy of partition for fault tolerance | 1 | 3 (min.insync.replicas=2) |
| ISR | In-Sync Replicas | All replicas | All replicas minus lagging ones |
| Consumer Group | Set of consumers sharing workload | N/A | One per microservice |
| Offset | Sequential message position | 0 | Auto-committed or manual |
| ZooKeeper | Cluster coordination service | Port 2181 | 3 or 5 node ensemble |
| Controller | Broker managing cluster metadata | Dynamic election | Dedicated controller broker |
| Segment | Physical log file | 1GB | Default (tunable) |
Code Examples
Creating a Topic with Advanced Configuration
# Create topic with specific configurations
kafka-topics.sh --create \
--bootstrap-server kafka-broker-0:9092,kafka-broker-1:9092,kafka-broker-2:9092 \
--topic order-events \
--partitions 12 \
--replication-factor 3 \
--config min.insync.replicas=2 \
--config max.message.bytes=10485760 \
--config retention.ms=604800000 \
--config cleanup.policy=delete \
--config compression.type=lz4 \
--config segment.bytes=1073741824 \
--config flush.messages=10000 \
--config index.interval.bytes=4096 \
--config unclean.leader.election.enable=false
Cluster Monitoring Script
#!/bin/bash
# Monitor Kafka cluster health and partition distribution
BOOTSTRAP_SERVER="kafka-broker-0:9092"
echo "=== KAFKA CLUSTER STATUS ==="
echo "Timestamp: $(date)"
echo ""
# List all brokers
echo "--- Active Brokers ---"
kafka-broker-api-versions.sh --bootstrap-server $BOOTSTRAP_SERVER | head -20
echo ""
echo "--- Topic Partition Distribution ---"
kafka-topics.sh --describe --bootstrap-server $BOOTSTRAP_SERVER --topic order-events
echo ""
echo "--- Consumer Group Lag ---"
kafka-consumer-groups.sh --bootstrap-server $BOOTSTRAP_SERVER \
--group order-processing-service \
--describe
echo ""
echo "--- ISR shrink/expand events ---"
kafka-server-stop.sh --force 2>/dev/null
Java Producer Configuration
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.Future;
public class KafkaProducerExample {
public static void main(String[] args) {
Properties props = new Properties();
// Core configurations
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"kafka-broker-0:9092,kafka-broker-1:9092,kafka-broker-2:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
// Reliability configurations
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 10);
props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 100);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);
// Idempotent producer for exactly-once semantics
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");
// Performance configurations
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 65536);
props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
// Monitoring
props.put(ProducerConfig.METRICS_NUM_SAMPLES_CONFIG, 10);
props.put(ProducerConfig.METRICS_SAMPLE_WINDOW_MS_CONFIG, 30000);
KafkaProducer<String, String> producer = new KafkaProducer<>(props);
// Initialize transactions
producer.initTransactions();
try {
producer.beginTransaction();
for (int i = 0; i < 1000; i++) {
String key = "order-" + (i % 100);
String value = "{\"orderId\": \"" + i + "\", \"amount\": " + (Math.random() * 1000) + "}";
ProducerRecord<String, String> record =
new ProducerRecord<>("order-events", key, value);
record.headers().add("correlation-id",
java.util.UUID.randomUUID().toString().getBytes());
record.headers().add("source-system", "order-service".getBytes());
Future<RecordMetadata> future = producer.send(record,
(metadata, exception) -> {
if (exception != null) {
System.err.println("Failed to send record: " + exception.getMessage());
} else {
System.out.printf("Sent record to partition %d, offset %d%n",
metadata.partition(), metadata.offset());
}
});
}
producer.commitTransaction();
} catch (Exception e) {
producer.abortTransaction();
throw e;
} finally {
producer.close();
}
}
}
Java Consumer with Advanced Configurations
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
public class KafkaConsumerExample {
private static final AtomicBoolean running = new AtomicBoolean(true);
public static void main(String[] args) {
Properties props = new Properties();
// Core configurations
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"kafka-broker-0:9092,kafka-broker-1:9092,kafka-broker-2:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processing-service");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
// Consumer behavior
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);
props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 45000);
props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 15000);
// Isolation level for transactional reads
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
// Partition assignment strategy
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG,
"org.apache.kafka.clients.consumer.CooperativeStickyAssignor");
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
// Subscribe to topics
consumer.subscribe(Arrays.asList("order-events", "payment-events"),
new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
System.out.println("Partitions revoked: " + partitions);
// Commit offsets before rebalance
consumer.commitSync(Duration.ofSeconds(30));
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
System.out.println("Partitions assigned: " + partitions);
}
});
try {
while (running.get()) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
processRecord(record);
}
// Commit offsets after successful processing
if (!records.isEmpty()) {
consumer.commitSync(Duration.ofSeconds(30));
}
}
} finally {
consumer.close();
}
}
private static void processRecord(ConsumerRecord<String, String> record) {
System.out.printf("Topic: %s, Partition: %d, Offset: %d, Key: %s, Value: %s%n",
record.topic(), record.partition(), record.offset(),
record.key(), record.value());
}
}
Performance Metrics
| Metric | Single Broker | 3-Broker Cluster | 6-Broker Cluster |
|---|---|---|---|
| Throughput (writes/sec) | 100K | 300K | 600K |
| Throughput (reads/sec) | 200K | 600K | 1.2M |
| Latency (p99) | 5ms | 10ms | 15ms |
| Latency (p999) | 15ms | 25ms | 40ms |
| Message Size (avg) | 1KB | 1KB | 1KB |
| Replication Lag | 0ms | <100ms | <200ms |
| Recovery Time | 0 | 30s | 60s |
| Disk Usage | 1TB | 3TB | 6TB |
| Memory (JVM Heap) | 6GB | 6GB | 6GB |
| Network Bandwidth | 1Gbps | 3Gbps | 6Gbps |
Best Practices
-
Partition Count Planning: Start with partitions = (target_throughput / broker_throughput) * replication_factor. Monitor consumer lag to determine if rebalancing is needed.
-
Replication Factor: Always use replication-factor ≥ 3 for production topics. Set
min.insync.replicas=2to ensure at least two replicas acknowledge writes before acknowledging to the producer. -
Consumer Group Sizing: Ensure number of consumers ≤ number of partitions. More consumers than partitions results in idle consumers. Use
CooperativeStickyAssignorfor incremental rebalancing to minimize pause time. -
Offset Management: Disable auto-commit (
enable.auto.commit=false) and commit offsets explicitly after successful processing to avoid data loss or duplicates. -
Broker Configuration: Use dedicated disks for Kafka data (avoid shared with OS). Configure
num.io.threadsandnum.network.threadsbased on CPU cores. Setlog.flush.interval.messagesandlog.flush.interval.mscarefully to balance durability with performance. -
Topic Configuration: Use
cleanup.policy=compactfor topics that need to retain latest state. Setretention.msandretention.bytesappropriately based on data volume and replay requirements. -
Monitoring: Monitor consumer lag, ISR shrink/expand events, under-replicated partitions, and disk usage. Use tools like Burrow, Cruise Control, or Confluent Control Center.
-
Security: Enable SSL/TLS for inter-broker and client-broker communication. Use SASL for authentication. Implement ACLs for authorization. Use separate listeners for internal and external traffic.
-
ZooKeeper/KRaft: For clusters > 100K partitions, consider migrating to KRaft mode to eliminate ZooKeeper dependency. Use dedicated ZooKeeper nodes (3 or 5) with sufficient I/O capacity.
-
Data Retention: Implement tiered storage for long-term retention. Use log compaction for topics that serve as materialized views. Archive to object storage (S3, GCS) for compliance requirements.
Key Takeaways:
- Kafka partitions are the unit of parallelism; throughput scales linearly with partition count up to broker limits
- ISR mechanism balances durability (acks=all) with availability; min.insync.replicas prevents unavailability during broker failures
- Consumer groups map partitions 1:1 to consumers for parallel consumption; beyond N consumers, extras are idle
- Ordering is guaranteed only within a partition; cross-partition ordering requires key-based routing
- Replication lag should be monitored and kept under replica.lag.time.max.ms to avoid ISR shrinkage
See also: Data Engineering Pipeline patterns (data-engineering/019) | PySpark Structured Streaming (pyspark/11-structured-streaming)