High-throughput, fault-tolerant messaging system for real-time data streaming and event-driven architectures
What is Apache Kafka?
Apache Kafka is a distributed event streaming platform designed for high-throughput, low-latency processing of real-time data feeds. It acts as a distributed commit log that can handle millions of messages per second.
Key Characteristics:
- Distributed: Runs on clusters of servers
- Fault-tolerant: Data replication across multiple nodes
- Scalable: Horizontal scaling by adding brokers
- Durable: Persistent storage with configurable retention
- High-throughput: Millions of messages per second
Kafka Architecture
Core Components
[Producer] → [Kafka Cluster] → [Consumer]
↓
[Topic] → [Partition] → [Replica]
↓ ↓ ↓
[Broker] → [Leader] → [Follower]
Detailed Architecture Diagram
Producer A ──┐
Producer B ──┼─→ [Topic: user-events]
Producer C ──┘ ├─ Partition 0 (Leader: Broker 1, Replicas: Broker 2,3)
├─ Partition 1 (Leader: Broker 2, Replicas: Broker 1,3)
└─ Partition 2 (Leader: Broker 3, Replicas: Broker 1,2)
↓
Consumer Group 1 ─────────┼─→ Consumer A
Consumer Group 2 ─────────┼─→ Consumer B
└─→ Consumer C
Core Components Deep Dive
1. Kafka Broker
# Kafka Broker Configuration
class KafkaBroker:
def __init__(self, broker_id, port=9092):
self.broker_id = broker_id
self.port = port
self.topics = {}
self.replica_manager = ReplicaManager()
self.log_manager = LogManager()
def handle_produce_request(self, topic, partition, messages):
# 1. Validate topic and partition
if not self.validate_topic_partition(topic, partition):
return {"error": "INVALID_TOPIC_PARTITION"}
# 2. Check if this broker is the leader for the partition
if not self.is_partition_leader(topic, partition):
return {"error": "NOT_LEADER_FOR_PARTITION"}
# 3. Append messages to log
offsets = self.log_manager.append_messages(topic, partition, messages)
# 4. Replication: followers FETCH (pull) from the leader; with acks=all the
# leader responds once all in-sync replicas (ISR) have the records
self.replica_manager.wait_for_isr_replication(
topic, partition, offsets
)
return {"status": "SUCCESS", "offsets": offsets}
def handle_fetch_request(self, topic, partition, offset, max_bytes):
# Fetch messages from the specified offset
messages = self.log_manager.fetch_messages(
topic, partition, offset, max_bytes
)
return {"messages": messages, "high_water_mark": self.get_high_water_mark(topic, partition)}2. Topics and Partitions
class KafkaTopic:
def __init__(self, name, num_partitions, replication_factor):
self.name = name
self.num_partitions = num_partitions
self.replication_factor = replication_factor
self.partitions = self._create_partitions()
def _create_partitions(self):
partitions = {}
for partition_id in range(self.num_partitions):
partitions[partition_id] = KafkaPartition(
topic=self.name,
partition_id=partition_id,
replication_factor=self.replication_factor
)
return partitions
def get_partition_for_key(self, key):
"""Determine which partition a message should go to based on key"""
if key is None:
# Sticky partitioning (default since Kafka 2.4): keep sending to one
# partition until its batch is full, then switch -> better batching
return self.current_sticky_partition()
else:
# Hash-based partitioning: Java client uses murmur2(key bytes), which is
# stable across processes (Python's hash() is randomised per process)
return murmur2(key) % self.num_partitions
class KafkaPartition:
def __init__(self, topic, partition_id, replication_factor):
self.topic = topic
self.partition_id = partition_id
self.replication_factor = replication_factor
self.leader = None
self.replicas = []
self.in_sync_replicas = [] # ISR
self.log = PartitionLog()
def assign_replicas(self, broker_list):
"""Assign partition replicas across brokers"""
start_index = self.partition_id % len(broker_list)
self.replicas = []
for i in range(self.replication_factor):
broker_index = (start_index + i) % len(broker_list)
self.replicas.append(broker_list[broker_index])
self.leader = self.replicas[0] # First replica is leader
self.in_sync_replicas = self.replicas.copy()3. Producer Implementation
class KafkaProducer:
def __init__(self, bootstrap_servers, **configs):
self.bootstrap_servers = bootstrap_servers
self.batch_size = configs.get('batch_size', 16384) # 16KB
self.linger_ms = configs.get('linger_ms', 5) # default 5ms since Kafka 4.0 (was 0)
self.acks = configs.get('acks', 'all') # default 'all' since Kafka 3.0
self.retries = configs.get('retries', 2147483647) # default Integer.MAX_VALUE since 2.1,
self.delivery_timeout_ms = configs.get('delivery_timeout_ms', 120000) # bounded by this (2 min)
self.enable_idempotence = configs.get('enable_idempotence', True) # default true since 3.0
self.compression_type = configs.get('compression_type', 'none')
self.metadata_cache = MetadataCache()
self.record_accumulator = RecordAccumulator()
self.network_client = NetworkClient()
def send(self, topic, value, key=None, partition=None, headers=None):
"""Send a message to Kafka"""
# 1. Get topic metadata
topic_metadata = self.metadata_cache.get_metadata(topic)
if not topic_metadata:
topic_metadata = self.fetch_metadata(topic)
# 2. Determine partition
if partition is None:
partition = self.get_partition(topic, key, topic_metadata)
# 3. Serialize message
serialized_record = ProducerRecord(
topic=topic,
partition=partition,
key=key,
value=value,
headers=headers,
timestamp=int(time.time() * 1000)
)
# 4. Add to batch
future = self.record_accumulator.append(serialized_record)
# 5. Send batch if conditions met
if self.should_send_batch(topic, partition):
self.send_batch(topic, partition)
return future
def send_batch(self, topic, partition):
"""Send accumulated batch to Kafka broker"""
batch = self.record_accumulator.get_batch(topic, partition)
if not batch:
return
# Compress batch if configured
if self.compression_type != 'none':
batch = self.compress_batch(batch, self.compression_type)
# Get leader broker for partition
leader_broker = self.metadata_cache.get_leader(topic, partition)
# Send to broker with retry logic
for attempt in range(self.retries + 1):
try:
response = self.network_client.send_produce_request(
broker=leader_broker,
topic=topic,
partition=partition,
batch=batch,
acks=self.acks
)
if response.status == "SUCCESS":
batch.complete_successfully(response.base_offset)
break
else:
if attempt == self.retries:
batch.complete_exceptionally(response.error)
else:
time.sleep(2 ** attempt) # Exponential backoff
except Exception as e:
if attempt == self.retries:
batch.complete_exceptionally(e)
else:
time.sleep(2 ** attempt)4. Consumer Implementation
class KafkaConsumer:
def __init__(self, topics, group_id, bootstrap_servers, **configs):
self.topics = topics
self.group_id = group_id
self.bootstrap_servers = bootstrap_servers
self.auto_offset_reset = configs.get('auto_offset_reset', 'latest')
self.enable_auto_commit = configs.get('enable_auto_commit', True)
self.auto_commit_interval_ms = configs.get('auto_commit_interval_ms', 5000)
self.max_poll_records = configs.get('max_poll_records', 500)
self.session_timeout_ms = configs.get('session_timeout_ms', 45000) # default 45s since Kafka 3.0 (was 10s)
self.coordinator = GroupCoordinator()
self.partition_assignor = PartitionAssignor()
self.offset_manager = OffsetManager()
def poll(self, timeout_ms=1000):
"""Poll for new messages"""
# 1. Ensure group membership
if not self.is_group_member():
self.join_group()
# 2. Handle rebalancing if needed
if self.needs_rebalance():
self.handle_rebalance()
# 3. Fetch messages from assigned partitions
messages = {}
for topic_partition in self.assigned_partitions:
partition_messages = self.fetch_from_partition(
topic_partition, timeout_ms
)
if partition_messages:
messages[topic_partition] = partition_messages
# 4. Update offsets
if self.enable_auto_commit:
self.maybe_auto_commit_offsets()
return messages
def join_group(self):
"""Join consumer group through group coordinator"""
# 1. Find group coordinator
coordinator_broker = self.find_coordinator(self.group_id)
# 2. Send join group request
join_response = coordinator_broker.send_join_group_request(
group_id=self.group_id,
member_id=self.member_id,
protocol_type="consumer",
group_protocols=["range", "roundrobin"]
)
# 3. Handle leader election
if join_response.leader_id == join_response.member_id:
# This consumer is the group leader
assignment = self.assign_partitions(join_response.members)
# Send sync group request with assignments (leader also gets its own
# assignment back in the SyncGroup response)
sync_response = self.coordinator.send_sync_group_request(
group_id=self.group_id,
generation_id=join_response.generation_id,
member_id=join_response.member_id,
group_assignment=assignment
)
self.assigned_partitions = sync_response.partition_assignment
else:
# Wait for assignment from leader
sync_response = self.coordinator.send_sync_group_request(
group_id=self.group_id,
generation_id=join_response.generation_id,
member_id=join_response.member_id
)
self.assigned_partitions = sync_response.partition_assignment
def assign_partitions(self, group_members):
"""Assign partitions to group members (if this consumer is leader)"""
all_partitions = []
# Get all partitions for subscribed topics
for topic in self.topics:
topic_metadata = self.get_topic_metadata(topic)
for partition_id in range(topic_metadata.partition_count):
all_partitions.append(TopicPartition(topic, partition_id))
# Use round-robin assignment strategy
assignment = {}
for i, partition in enumerate(all_partitions):
member_index = i % len(group_members)
member_id = group_members[member_index].member_id
if member_id not in assignment:
assignment[member_id] = []
assignment[member_id].append(partition)
return assignmentMessage Delivery Semantics
1. At-Least-Once Delivery
# Producer configuration for at-least-once
producer_config = {
'acks': 'all', # Wait for all replicas to acknowledge
'retries': 3, # Retry failed sends
'enable_idempotence': False
}
# Consumer configuration
consumer_config = {
'enable_auto_commit': False, # Manual offset management
'isolation_level': 'read_committed'
}
class AtLeastOnceConsumer:
def process_messages(self):
while True:
records = self.consumer.poll(timeout_ms=1000)
for topic_partition, messages in records.items():
for message in messages:
try:
# Process message
self.process_message(message)
# Commit offset after successful processing
self.consumer.commit_sync({
topic_partition: OffsetAndMetadata(message.offset + 1)
})
except Exception as e:
# Don't commit offset on failure - message will be redelivered
self.logger.error(f"Failed to process message: {e}")2. Exactly-Once Delivery
# Note: Kafka EOS covers read-process-write WITHIN Kafka (consume → produce → offsets).
# Side effects in external systems (DB writes, payments, emails) still need
# idempotency keys / upserts / outbox to avoid duplicates.
# Producer configuration for exactly-once
producer_config = {
'enable_idempotence': True,
'acks': 'all',
'retries': Integer.MAX_VALUE,
'max_in_flight_requests_per_connection': 5,
'transactional_id': 'my-transactional-producer'
}
class ExactlyOnceProcessor:
def __init__(self):
self.producer = KafkaProducer(**producer_config)
self.consumer = KafkaConsumer(**consumer_config)
self.producer.init_transactions()
def process_with_exactly_once(self):
while True:
records = self.consumer.poll(timeout_ms=1000)
if not records:
continue
# Begin transaction
self.producer.begin_transaction()
try:
for topic_partition, messages in records.items():
for message in messages:
# Process message
result = self.process_message(message)
# Send result to output topic within transaction
self.producer.send('output-topic', result)
# Commit consumer offsets within transaction
offsets = self.get_current_offsets(records)
self.producer.send_offsets_to_transaction(
offsets, self.consumer.group_id
)
# Commit transaction
self.producer.commit_transaction()
except Exception as e:
# Abort transaction on error
self.producer.abort_transaction()
self.logger.error(f"Transaction aborted: {e}")3. At-Most-Once Delivery
# Producer configuration for at-most-once
producer_config = {
'acks': '0', # No acknowledgment required
'retries': 0 # No retries
}
class AtMostOnceConsumer:
def process_messages(self):
while True:
records = self.consumer.poll(timeout_ms=1000)
# Commit the offsets of THIS batch before processing it
# (committing before poll() would commit the previous batch → at-least-once)
self.consumer.commit_sync()
for topic_partition, messages in records.items():
for message in messages:
try:
self.process_message(message)
except Exception as e:
# Message is lost on failure since offset was already committed
self.logger.error(f"Message lost due to processing error: {e}")Kafka Zookeeper & KRaft
Traditional Zookeeper Architecture
class ZookeeperCoordinator:
def __init__(self, zk_hosts):
self.zk_client = KazooClient(hosts=zk_hosts)
self.zk_client.start()
def register_broker(self, broker_id, host, port):
"""Register broker with Zookeeper"""
broker_path = f"/brokers/ids/{broker_id}"
broker_data = {
"host": host,
"port": port,
"timestamp": int(time.time() * 1000)
}
self.zk_client.create(
broker_path,
json.dumps(broker_data).encode(),
ephemeral=True
)
def elect_controller(self, broker_id):
"""Elect cluster controller"""
controller_path = "/controller"
controller_data = {
"brokerid": broker_id,
"timestamp": int(time.time() * 1000)
}
try:
self.zk_client.create(
controller_path,
json.dumps(controller_data).encode(),
ephemeral=True
)
return True # This broker is now controller
except NodeExistsError:
return False # Another broker is controller
def watch_broker_changes(self, callback):
"""Watch for broker additions/removals"""
self.zk_client.ChildrenWatch("/brokers/ids", callback)KRaft (Kafka Raft) - Zookeeper Replacement
class KRaftController:
def __init__(self, node_id, controller_nodes):
self.node_id = node_id
self.controller_nodes = controller_nodes
self.raft_state = RaftState()
self.metadata_log = MetadataLog()
self.current_term = 0
self.voted_for = None
self.state = "FOLLOWER" # LEADER, FOLLOWER, CANDIDATE
def start_election(self):
"""Start leader election using Raft protocol"""
self.state = "CANDIDATE"
self.current_term += 1
self.voted_for = self.node_id
votes_received = 1 # Vote for self
# Request votes from other controllers
for node in self.controller_nodes:
if node.node_id != self.node_id:
response = node.request_vote(
term=self.current_term,
candidate_id=self.node_id,
last_log_index=self.metadata_log.last_index(),
last_log_term=self.metadata_log.last_term()
)
if response.vote_granted:
votes_received += 1
# Become leader if majority votes received
if votes_received > len(self.controller_nodes) // 2:
self.become_leader()
def become_leader(self):
"""Become cluster leader"""
self.state = "LEADER"
self.logger.info(f"Node {self.node_id} became leader for term {self.current_term}")
# Start sending heartbeats
self.start_heartbeat_timer()
# Handle metadata changes
self.start_metadata_management()
def append_metadata_record(self, record):
"""Append metadata change to log and replicate"""
if self.state != "LEADER":
raise Exception("Only leader can append metadata records")
# Append to local log
log_entry = LogEntry(
term=self.current_term,
index=self.metadata_log.next_index(),
record=record
)
self.metadata_log.append(log_entry)
# Replicate to followers
for follower in self.get_followers():
self.replicate_to_follower(follower, log_entry)Performance Optimizations
1. Zero-Copy Optimization
class ZeroCopyOptimization:
"""
Kafka uses zero-copy optimization to transfer data efficiently
from disk to network without copying data through user space
"""
def send_file_with_zero_copy(self, socket, file_path, offset, count):
"""
Use sendfile() system call for zero-copy data transfer
Data flows: Disk → Kernel Buffer → Network Card
Bypasses: User Space copying
"""
with open(file_path, 'rb') as file:
# Linux sendfile() - zero-copy transfer
socket.sendfile(file, offset, count)
def traditional_copy_vs_zero_copy(self):
"""
Traditional Copy (4 context switches, 2 data copies):
1. read() - Disk → Kernel Buffer → User Buffer
2. write() - User Buffer → Kernel Buffer → Network
Zero Copy (2 context switches, 0 CPU copies - only DMA copies):
1. sendfile() - Disk → Kernel Buffer → Network
Performance Impact (workload-dependent, no fixed figure):
- Lower CPU usage
- Higher throughput
- Lower memory utilization
- Not applicable with TLS: broker must encrypt data in user space
"""
pass2. Batch Processing & Compression
class KafkaOptimizations:
def __init__(self):
self.batch_configurations = {
'batch_size': 16384, # 16KB batches
'linger_ms': 5, # Wait 5ms for more messages
'compression_type': 'lz4' # Fast compression
}
def optimize_producer_batching(self):
"""
Producer batching strategies for optimal throughput
"""
return {
# High Throughput Configuration
'high_throughput': {
'batch_size': 65536, # 64KB
'linger_ms': 20, # Wait longer for larger batches
'compression_type': 'lz4', # Fast compression
'buffer_memory': 67108864, # 64MB buffer
'acks': '1' # Leader acknowledgment only
},
# Low Latency Configuration
'low_latency': {
'batch_size': 1024, # 1KB
'linger_ms': 0, # Send immediately
'compression_type': 'none', # No compression overhead
'acks': '1'
},
# Balanced Configuration
'balanced': {
'batch_size': 16384, # 16KB
'linger_ms': 5, # 5ms wait
'compression_type': 'snappy', # Good compression ratio
'acks': 'all' # Full replication
}
}
def compression_comparison(self):
"""
Compression algorithms comparison for Kafka
"""
return {
'none': {
'compression_ratio': 1.0,
'cpu_usage': 'lowest',
'throughput': 'highest',
'use_case': 'low latency requirements'
},
'gzip': {
'compression_ratio': 4.0,
'cpu_usage': 'high',
'throughput': 'low',
'use_case': 'network bandwidth limited'
},
'snappy': {
'compression_ratio': 2.5,
'cpu_usage': 'medium',
'throughput': 'high',
'use_case': 'balanced performance'
},
'lz4': {
'compression_ratio': 2.0,
'cpu_usage': 'low',
'throughput': 'highest',
'use_case': 'high throughput scenarios'
},
'zstd': {
'compression_ratio': 3.5,
'cpu_usage': 'medium',
'throughput': 'medium',
'use_case': 'best compression with good speed'
}
}3. Partitioning Strategies
class PartitioningStrategies:
def __init__(self, num_partitions):
self.num_partitions = num_partitions
def key_based_partitioning(self, key):
"""
Hash-based partitioning ensures messages with same key
go to same partition (maintaining order)
Kafka's Java client uses murmur2 (stable across processes), not Python hash()
"""
return murmur2(key) % self.num_partitions
def round_robin_partitioning(self):
"""
Distribute messages evenly across all partitions
Good for load balancing when order doesn't matter
"""
self.current_partition = getattr(self, 'current_partition', -1)
self.current_partition = (self.current_partition + 1) % self.num_partitions
return self.current_partition
def custom_partitioning(self, message):
"""
Custom business logic for partitioning
Example: Geographic partitioning
"""
if hasattr(message, 'region'):
region_mapping = {
'us-east': [0, 1, 2],
'us-west': [3, 4, 5],
'europe': [6, 7, 8],
'asia': [9, 10, 11]
}
region_partitions = region_mapping.get(message.region, [0])
return random.choice(region_partitions)
return self.round_robin_partitioning()
def optimal_partition_count(self, target_throughput_mb_per_sec):
"""
Calculate optimal partition count based on throughput requirements
Rule of thumb:
- Each partition can handle ~10-100 MB/sec
- Consider consumer parallelism needs
- Account for replication overhead
"""
partition_throughput_mb = 50 # Conservative estimate
recommended_partitions = max(
1,
int(target_throughput_mb_per_sec / partition_throughput_mb)
)
# Round up to next power of 2 for better distribution
return 2 ** (recommended_partitions - 1).bit_length()Use Cases & Patterns
1. Event Sourcing Pattern
class EventSourcingWithKafka:
def __init__(self):
self.event_store = KafkaProducer(topic='events')
self.snapshot_store = SnapshotStore()
def handle_command(self, aggregate_id, command):
"""Process command and generate events"""
# 1. Load aggregate from events
aggregate = self.load_aggregate(aggregate_id)
# 2. Process command
events = aggregate.handle_command(command)
# 3. Store events in Kafka
for event in events:
self.event_store.send(
key=aggregate_id,
value=event,
headers={'event_type': event.__class__.__name__}
)
# 4. Update aggregate version
aggregate.version += len(events)
return events
def load_aggregate(self, aggregate_id):
"""Reconstruct aggregate from events"""
# Try to load from snapshot first
snapshot = self.snapshot_store.get_latest_snapshot(aggregate_id)
if snapshot:
aggregate = snapshot.aggregate
from_version = snapshot.version
else:
aggregate = self.create_empty_aggregate(aggregate_id)
from_version = 0
# Replay events from snapshot version
events = self.get_events_from_version(aggregate_id, from_version)
for event in events:
aggregate.apply_event(event)
return aggregate
# Example: Bank Account Event Sourcing
class BankAccountAggregate:
def __init__(self, account_id):
self.account_id = account_id
self.balance = 0
self.version = 0
self.is_closed = False
def handle_command(self, command):
if command.type == "DEPOSIT":
return [DepositEvent(self.account_id, command.amount)]
elif command.type == "WITHDRAW":
if self.balance >= command.amount:
return [WithdrawEvent(self.account_id, command.amount)]
else:
raise InsufficientFundsError()
elif command.type == "CLOSE_ACCOUNT":
return [AccountClosedEvent(self.account_id)]
def apply_event(self, event):
if isinstance(event, DepositEvent):
self.balance += event.amount
elif isinstance(event, WithdrawEvent):
self.balance -= event.amount
elif isinstance(event, AccountClosedEvent):
self.is_closed = True2. CQRS (Command Query Responsibility Segregation)
class CQRSWithKafka:
def __init__(self):
self.command_handler = CommandHandler()
self.event_stream = KafkaProducer(topic='events')
self.read_model_updater = ReadModelUpdater()
def setup_cqrs_pipeline(self):
"""Setup CQRS pipeline with Kafka"""
# Command side
@self.command_handler.route("CREATE_ORDER")
def handle_create_order(command):
# Business logic
order = Order.create(command.customer_id, command.items)
# Generate events
events = [
OrderCreatedEvent(order.id, order.customer_id, order.items),
InventoryReservedEvent(order.id, order.items)
]
# Publish events
for event in events:
self.event_stream.send(
key=order.id,
value=event
)
return order.id
# Query side - Event handlers for read models
@self.read_model_updater.handle("OrderCreatedEvent")
def update_order_view(event):
order_view = {
'order_id': event.order_id,
'customer_id': event.customer_id,
'status': 'CREATED',
'items': event.items,
'created_at': event.timestamp
}
self.order_view_db.insert(order_view)
@self.read_model_updater.handle("InventoryReservedEvent")
def update_inventory_view(event):
for item in event.items:
self.inventory_view_db.decrement_available(
item.product_id,
item.quantity
)
class ReadModelUpdater:
def __init__(self):
self.event_consumer = KafkaConsumer(
topics=['events'],
group_id='read-model-updater'
)
def start_processing(self):
"""Start processing events to update read models"""
while True:
messages = self.event_consumer.poll(timeout_ms=1000)
for topic_partition, message_list in messages.items():
for message in message_list:
event = self.deserialize_event(message.value)
handler = self.get_handler(event.__class__.__name__)
if handler:
try:
handler(event)
self.event_consumer.commit()
except Exception as e:
self.logger.error(f"Failed to process event: {e}")
# Implement retry logic or dead letter queue3. Microservices Communication
class MicroservicesCommunication:
def __init__(self):
self.event_bus = KafkaProducer()
self.command_bus = KafkaProducer()
def setup_saga_pattern(self):
"""Distributed transaction using Saga pattern"""
class OrderSaga:
def __init__(self):
self.state = "STARTED"
self.compensations = []
def handle_order_created(self, event):
# Step 1: Reserve inventory
self.command_bus.send(
topic='inventory-commands',
value=ReserveInventoryCommand(
order_id=event.order_id,
items=event.items
)
)
self.compensations.append(
ReleaseInventoryCommand(event.order_id)
)
def handle_inventory_reserved(self, event):
# Step 2: Process payment
self.command_bus.send(
topic='payment-commands',
value=ProcessPaymentCommand(
order_id=event.order_id,
amount=event.total_amount
)
)
self.compensations.append(
RefundPaymentCommand(event.order_id)
)
def handle_payment_processed(self, event):
# Step 3: Ship order
self.command_bus.send(
topic='shipping-commands',
value=ShipOrderCommand(
order_id=event.order_id,
address=event.shipping_address
)
)
self.state = "COMPLETED"
def handle_payment_failed(self, event):
# Compensate: Release inventory
for compensation in reversed(self.compensations):
self.command_bus.send(
topic=compensation.topic,
value=compensation
)
self.state = "COMPENSATED"Monitoring & Operations
Kafka Metrics & KPIs
class KafkaMonitoring:
def __init__(self):
self.metrics_collector = MetricsCollector()
self.alert_manager = AlertManager()
def collect_broker_metrics(self):
"""Collect essential broker metrics"""
return {
# Throughput Metrics
'messages_per_second': self.get_metric('kafka.broker.messages.in.rate'),
'bytes_per_second_in': self.get_metric('kafka.broker.bytes.in.rate'),
'bytes_per_second_out': self.get_metric('kafka.broker.bytes.out.rate'),
# Latency Metrics
'request_latency_ms': self.get_metric('kafka.request.latency.avg'),
'produce_latency_ms': self.get_metric('kafka.broker.produce.latency.avg'),
'fetch_latency_ms': self.get_metric('kafka.broker.fetch.latency.avg'),
# Partition Metrics
'under_replicated_partitions': self.get_metric('kafka.broker.under.replicated.partitions'),
'offline_partitions': self.get_metric('kafka.broker.offline.partitions'),
'leader_election_rate': self.get_metric('kafka.broker.leader.election.rate'),
# Resource Metrics
'cpu_usage_percent': self.get_metric('system.cpu.usage'),
'memory_usage_percent': self.get_metric('system.memory.usage'),
'disk_usage_percent': self.get_metric('system.disk.usage'),
'network_io': self.get_metric('system.network.io'),
# Consumer Lag
'consumer_lag': self.get_consumer_lag_metrics()
}
def setup_alerts(self):
"""Setup critical alerts for Kafka cluster"""
alerts = [
Alert(
name="High Consumer Lag",
condition="consumer_lag > 1000000", # 1M messages behind
severity="CRITICAL",
action="scale_consumers"
),
Alert(
name="Under Replicated Partitions",
condition="under_replicated_partitions > 0",
severity="HIGH",
action="check_broker_health"
),
Alert(
name="High Request Latency",
condition="request_latency_ms > 1000", # > 1 second
severity="MEDIUM",
action="investigate_performance"
),
Alert(
name="Offline Partitions",
condition="offline_partitions > 0",
severity="CRITICAL",
action="immediate_investigation"
)
]
for alert in alerts:
self.alert_manager.register_alert(alert)
def health_check_dashboard(self):
"""Create health check dashboard"""
return {
'cluster_status': self.get_cluster_status(),
'broker_health': self.get_broker_health(),
'topic_health': self.get_topic_health(),
'consumer_group_health': self.get_consumer_group_health(),
'performance_metrics': self.get_performance_summary(),
'capacity_planning': self.get_capacity_metrics()
}Performance Tuning Guidelines
| Configuration | Description | Recommended Value | Use Case |
|---|---|---|---|
| num.network.threads | Network threads | 8 | High throughput |
| num.io.threads | I/O threads | 16 | High I/O load |
| socket.send.buffer.bytes | Send buffer | 102400 (100KB) | High throughput |
| socket.receive.buffer.bytes | Receive buffer | 102400 (100KB) | High throughput |
| log.segment.bytes | Segment size | 1073741824 (1GB) | Large messages |
| log.retention.hours | Retention time | 168 (7 days) | Standard retention |
| compression.type | Compression | producer | Let producer decide |
| min.insync.replicas | Min replicas | 2 | High availability |
This comprehensive Apache Kafka guide covers all essential aspects for system design interviews and production usage, from basic concepts to advanced optimization techniques and real-world use cases.