🧠 Kafka Fundamentals
🎯 What is Apache Kafka?
Apache Kafka is a distributed event streaming platform. It's a publish-subscribe messaging system that stores streams of records in a fault-tolerant, durable way. It's used for real-time data pipelines, streaming analytics, and event-driven architectures.
Why Kafka Matters
Throughput: Millions of messages/sec. Durability: Persistent storage. Scalability: Horizontally scalable. Real-time: Low latency. Ecosystem: Connect, Streams. Foundation: Event-driven architecture.
📌 Publish-Subscribe📌 Distributed Log📌 Fault-Tolerant📌 High Throughput
# Apache Kafka — the big picture
# Kafka is a distributed event streaming platform.
# CONCEPTUAL OVERVIEW
print("""
=== Apache Kafka Overview ===
Kafka is:
- Publish-subscribe messaging system
- Distributed commit log
- Event streaming platform
- Fault-tolerant
- High throughput
Key concepts:
- Producer: Publishes messages
- Consumer: Subscribes to messages
- Topic: Category of messages
- Partition: Ordered log
- Broker: Kafka server
- Cluster: Multiple brokers
- Offset: Position in partition
- Consumer Group: Set of consumers
Use cases:
- Real-time data pipelines
- Streaming analytics
- Event-driven microservices
- Log aggregation
- Messaging
- Activity tracking
""")
# SIMPLE PRODUCER/CONSUMER EXAMPLE
print("=== Simple Producer (Python) ===")
print("""
from kafka import KafkaProducer
import json
# Create producer
producer = KafkaProducer(
bootstrap_servers=['localhost:9092'],
value_serializer=lambda v: json.dumps(v).encode('utf-8')
)
# Send messages
for i in range(10):
message = {'id': i, 'message': f'Hello {i}'}
producer.send('my-topic', message)
print(f"Sent: {message}")
# Flush and close
producer.flush()
producer.close()
""")
print("\n=== Simple Consumer (Python) ===")
print("""
from kafka import KafkaConsumer
import json
# Create consumer
consumer = KafkaConsumer(
'my-topic',
bootstrap_servers=['localhost:9092'],
auto_offset_reset='earliest',
group_id='my-group',
value_deserializer=lambda v: json.loads(v.decode('utf-8'))
)
# Consume messages
for message in consumer:
print(f"Received: {message.value}")
""")
# KEY PROPERTIES
print("""
=== Key Properties of Kafka ===
1. DISTRIBUTED
- Data spread across brokers
- Partitions for parallelism
- Fault-tolerant
2. PERSISTENT
- Messages stored on disk
- Configurable retention
- Replay capability
3. HIGH THROUGHPUT
- Millions of messages/sec
- Sequential I/O
- Batching
4. SCALABLE
- Add brokers
- Add partitions
- Linear scale
5. FAULT-TOLERANT
- Replication
- Automatic failover
- No data loss
6. REAL-TIME
- Low latency (ms)
- Streaming
- Event-driven
7. ECOSYSTEM
- Kafka Connect (integration)
- Kafka Streams (processing)
- KSQL (SQL on streams)
""")
# REAL-WORLD SCALE
print("=== Real-World Kafka Deployments ===")
deployments = [
('LinkedIn', '7+ trillion messages/day', 'Original creator'),
('Netflix', 'Petabytes/day', 'Recommendations, logging'),
('Uber', 'Trillions of messages/day', 'Real-time analytics'),
('Airbnb', 'Billions of messages/day', 'Event pipeline'),
('Twitter', 'Billions of messages/day', 'Timeline, analytics'),
('Goldman Sachs', 'Millions of messages/sec', 'Trading'),
]
print(f"{'Company':<15} {'Scale':<30} {'Use Case':<25}")
print("-" * 70)
for company, scale, use_case in deployments:
print(f"{company:<15} {scale:<30} {use_case:<25}")🏗️ Architecture
🎯 Kafka Architecture
Kafka's architecture consists of producers, consumers, brokers, and ZooKeeper (or KRaft). Data is organized into topics split into partitions.
Architecture Components
Producer: Publishes messages. Consumer: Subscribes to messages. Broker: Kafka server. Topic: Message category. Partition: Ordered log. ZooKeeper: Coordination.
🏗️ Producer🏗️ Consumer🏗️ Broker🏗️ Topic
# Kafka Architecture
print("=== Kafka Architecture Diagram ===")
print("""
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Producer 1 │ │ Producer 2 │ │ Producer 3 │
└──────┬──────┘ └──────┬──────┘ └──────┬──────┘
│ │ │
└────────────────┼────────────────┘
│
▼
┌─────────────────────────────────────────────────┐
│ KAFKA CLUSTER │
│ │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────┐ │
│ │ Broker 1 │ │ Broker 2 │ │Broker 3 │ │
│ │ │ │ │ │ │ │
│ │ Topic A P0 │ │ Topic A P1 │ │Topic A │ │
│ │ Topic B P0 │ │ Topic B P1 │ │ P2 │ │
│ └─────────────┘ └─────────────┘ └─────────┘ │
│ │
└─────────────────────┬───────────────────────────┘
│
▼
┌─────────────┐ ┌─────────────┐ ┌─────────────┐
│ Consumer G1 │ │ Consumer G1 │ │ Consumer G2 │
│ (C1) │ │ (C2) │ │ (C1) │
└─────────────┘ └─────────────┘ └─────────────┘
""")
# COMPONENTS
print("\n=== Kafka Components ===")
print("""
1. PRODUCER
- Publishes messages to topics
- Chooses partition
- Batching, compression
- Acknowledgment modes
2. CONSUMER
- Subscribes to topics
- Reads messages
- Tracks offset
- Consumer groups
3. BROKER
- Kafka server
- Stores messages
- Serves producers/consumers
- Replication
4. TOPIC
- Category of messages
- Logical name
- Multiple partitions
5. PARTITION
- Ordered log
- Immutable
- Offset per message
6. ZOOKEEPER (or KRaft)
- Cluster coordination
- Broker metadata
- Leader election
- Configuration
7. CONSUMER GROUP
- Set of consumers
- Partition assignment
- Parallel consumption
""")
# KRAFT vs ZOOKEEPER
print("\n=== KRaft vs ZooKeeper ===")
print(f"{'Aspect':<25} {'ZooKeeper':<25} {'KRaft':<25}")
print("-" * 75)
print(f"{'Coordination':<25} {'External':<25} {'Built-in':<25}")
print(f"{'Complexity':<25} {'Higher':<25} {'Lower':<25}")
print(f"{'Deployment':<25} {'Separate cluster':<25} {'Single cluster':<25}")
print(f"{'Scalability':<25} {'Limited':<25} {'Better':<25}")
print(f"{'Status':<25} {'Legacy':<25} {'Modern (Kafka 3.x+)':<25}")
print(f"{'Future':<25} {'Deprecated':<25} {'Recommended':<25}")
# EXECUTION FLOW
print("\n=== Message Flow ===")
print("""
1. PRODUCER SENDS
- Producer connects to broker
- Sends message to topic
- Chooses partition (key hash, round-robin)
2. BROKER STORES
- Appends to partition log
- Assigns offset
- Replicates to followers
3. CONSUMER READS
- Consumer subscribes to topic
- Assigned partitions
- Reads messages in order
- Commits offset
4. CONSUMER GROUP
- Partitions distributed
- Each partition to one consumer
- Parallel consumption
""")
# IMPLEMENTATION
print("\n=== Simulating Kafka ===")
import time
import threading
from collections import deque, defaultdict
import random
class KafkaBroker:
"""Simulate Kafka Broker."""
def __init__(self, broker_id):
self.id = broker_id
self.topics = {} # topic -> {partition -> messages}
self.offsets = defaultdict(int)
def create_topic(self, topic, num_partitions=3):
"""Create a topic."""
self.topics[topic] = {
i: [] for i in range(num_partitions)
}
print(f"Broker {self.id}: Created topic '{topic}' with {num_partitions} partitions")
def produce(self, topic, partition, message):
"""Produce a message."""
if topic not in self.topics:
print(f"Broker {self.id}: Topic '{topic}' not found")
return None
if partition not in self.topics[topic]:
print(f"Broker {self.id}: Partition {partition} not found")
return None
offset = len(self.topics[topic][partition])
self.topics[topic][partition].append({
'offset': offset,
'message': message,
'timestamp': time.time(),
})
return offset
def consume(self, topic, partition, offset):
"""Consume a message."""
if topic not in self.topics:
return None
if partition not in self.topics[topic]:
return None
messages = self.topics[topic][partition]
if offset < len(messages):
return messages[offset]
return None
def get_partition_count(self, topic):
"""Get number of partitions."""
if topic in self.topics:
return len(self.topics[topic])
return 0
def get_message_count(self, topic, partition):
"""Get message count."""
if topic in self.topics and partition in self.topics[topic]:
return len(self.topics[topic][partition])
return 0
class KafkaProducer:
"""Simulate Kafka Producer."""
def __init__(self, broker, serializer=None):
self.broker = broker
self.serializer = serializer or (lambda x: x)
self.sent_count = 0
def send(self, topic, message, key=None):
"""Send a message."""
# Choose partition
num_partitions = self.broker.get_partition_count(topic)
if num_partitions == 0:
print(f"Producer: Topic '{topic}' not found")
return None
if key is not None:
partition = hash(key) % num_partitions
else:
partition = self.sent_count % num_partitions
serialized = self.serializer(message)
offset = self.broker.produce(topic, partition, serialized)
self.sent_count += 1
return offset
class KafkaConsumer:
"""Simulate Kafka Consumer."""
def __init__(self, broker, topic, group_id, deserializer=None):
self.broker = broker
self.topic = topic
self.group_id = group_id
self.deserializer = deserializer or (lambda x: x)
self.offsets = defaultdict(int)
self.consumed_count = 0
def poll(self, partition):
"""Poll for messages."""
offset = self.offsets[partition]
message = self.broker.consume(self.topic, partition, offset)
if message:
self.offsets[partition] += 1
self.consumed_count += 1
return {
'partition': partition,
'offset': message['offset'],
'value': self.deserializer(message['message']),
}
return None
def poll_all(self):
"""Poll all partitions."""
messages = []
num_partitions = self.broker.get_partition_count(self.topic)
for partition in range(num_partitions):
msg = self.poll(partition)
if msg:
messages.append(msg)
return messages
# Simulate
print("=== Kafka Simulation ===")
# Create broker
broker = KafkaBroker('broker-1')
# Create topic
broker.create_topic('orders', num_partitions=3)
# Create producer
producer = KafkaProducer(broker, serializer=lambda x: x)
# Send messages
print("\nProducing messages:")
messages = [
{'order_id': 1, 'user': 'Alice', 'amount': 100},
{'order_id': 2, 'user': 'Bob', 'amount': 200},
{'order_id': 3, 'user': 'Charlie', 'amount': 150},
{'order_id': 4, 'user': 'Diana', 'amount': 300},
{'order_id': 5, 'user': 'Eve', 'amount': 250},
{'order_id': 6, 'user': 'Frank', 'amount': 175},
]
for msg in messages:
offset = producer.send('orders', msg, key=msg['user'])
print(f" Sent {msg['order_id']} → partition {hash(msg['user']) % 3}, offset {offset}")
# Show partition contents
print("\nPartition contents:")
for partition, msgs in broker.topics['orders'].items():
print(f" Partition {partition}:")
for msg in msgs:
print(f" offset {msg['offset']}: {msg['message']}")
# Create consumer
consumer = KafkaConsumer(broker, 'orders', 'order-processor')
# Consume messages
print("\nConsuming messages:")
for _ in range(10):
messages = consumer.poll_all()
if not messages:
break
for msg in messages:
print(f" Partition {msg['partition']}, offset {msg['offset']}: {msg['value']}")
print(f"\nTotal consumed: {consumer.consumed_count}")
# ARCHITECTURE SUMMARY
print("\n=== Architecture Summary ===")
print(f"{'Component':<25} {'Role':<40}")
print("-" * 65)
print(f"{'Producer':<25} {'Publishes messages':<40}")
print(f"{'Consumer':<25} {'Subscribes to messages':<40}")
print(f"{'Broker':<25} {'Kafka server':<40}")
print(f"{'Topic':<25} {'Message category':<40}")
print(f"{'Partition':<25} {'Ordered log':<40}")
print(f"{'ZooKeeper/KRaft':<25} {'Coordination':<40}")
print(f"{'Consumer Group':<25} {'Parallel consumption':<40}")📋 Topics & Partitions
🎯 Organizing Messages
Topics are categories of messages. Each topic is split into partitions — ordered, immutable logs that enable parallelism and scalability.
Topics & Partitions
Topic: Message category. Partition: Ordered log. Offset: Position. Ordering: Per partition. Parallelism: One consumer per partition. Retention: Time/size based.
📋 Topic📋 Partition📋 Offset📋 Ordering
# Topics and Partitions
print("=== What is a Topic? ===")
print("""
Topic:
- Category of messages
- Logical name (e.g., "orders")
- Multiple partitions
- Configurable retention
- Multiple producers/consumers
Example:
Topic: orders
Partitions: 3
Replication: 2
Messages distributed across partitions.
""")
print("\n=== What is a Partition? ===")
print("""
Partition:
- Ordered, immutable log
- Appended by producers
- Read by consumers
- Each message has an offset
- Ordering guaranteed per partition
Structure:
Partition 0: [msg0, msg1, msg2, msg3, ...]
↑
offset
Partition 1: [msg0, msg1, msg2, ...]
Partition 2: [msg0, msg1, ...]
""")
print("\n=== Partitioning Strategies ===")
print("""
1. KEY-BASED
- Hash(key) % num_partitions
- Same key → same partition
- Ordering per key
producer.send('topic', key='user123', value='data')
2. ROUND-ROBIN
- No key
- Distribute evenly
- No ordering guarantee
producer.send('topic', value='data')
3. CUSTOM PARTITIONER
- User-defined
- Business logic
class MyPartitioner(Partitioner):
def partition(self, key, partitions):
return ...
""")
print("\n=== Topic Configuration ===")
print("""
Key configurations:
retention.ms:
- How long to keep messages
- Default: 7 days (604800000 ms)
- -1 = infinite
retention.bytes:
- Max size per partition
- Default: -1 (infinite)
cleanup.policy:
- delete: Remove old messages
- compact: Keep latest per key
segment.bytes:
- Segment file size
- Default: 1 GB
min.insync.replicas:
- Min replicas for write
- Default: 1
- Recommended: 2
""")
# IMPLEMENTATION
print("\n=== Simulating Topics & Partitions ===")
class KafkaTopic:
"""Simulate Kafka Topic."""
def __init__(self, name, num_partitions=3, replication_factor=1, retention_ms=604800000):
self.name = name
self.num_partitions = num_partitions
self.replication_factor = replication_factor
self.retention_ms = retention_ms
self.partitions = [KafkaPartition(i) for i in range(num_partitions)]
self.config = {
'retention.ms': retention_ms,
'cleanup.policy': 'delete',
'segment.bytes': 1073741824,
}
def get_partition(self, key=None):
"""Get partition for a key."""
if key is None:
# Round-robin (simplified)
import random
return random.randint(0, self.num_partitions - 1)
return hash(key) % self.num_partitions
def produce(self, message, key=None):
"""Produce a message."""
partition_id = self.get_partition(key)
partition = self.partitions[partition_id]
offset = partition.append(message)
return partition_id, offset
def consume(self, partition_id, offset):
"""Consume a message."""
if 0 <= partition_id < self.num_partitions:
return self.partitions[partition_id].read(offset)
return None
def get_info(self):
"""Get topic info."""
return {
'name': self.name,
'partitions': self.num_partitions,
'replication': self.replication_factor,
'total_messages': sum(p.size() for p in self.partitions),
'config': self.config,
}
class KafkaPartition:
"""Simulate Kafka Partition."""
def __init__(self, partition_id):
self.id = partition_id
self.log = [] # List of messages
self.start_offset = 0
def append(self, message):
"""Append a message."""
offset = self.start_offset + len(self.log)
self.log.append({
'offset': offset,
'message': message,
'timestamp': time.time(),
})
return offset
def read(self, offset):
"""Read a message."""
for msg in self.log:
if msg['offset'] == offset:
return msg
return None
def read_range(self, start, end):
"""Read a range of messages."""
return [m for m in self.log if start <= m['offset'] < end]
def size(self):
"""Number of messages."""
return len(self.log)
def latest_offset(self):
"""Latest offset."""
return self.start_offset + len(self.log) - 1 if self.log else self.start_offset - 1
# Demo
print("=== Topic & Partition Demo ===")
# Create topic
topic = KafkaTopic('orders', num_partitions=3)
info = topic.get_info()
print(f"Topic: {info['name']}")
print(f"Partitions: {info['partitions']}")
print(f"Replication: {info['replication']}")
# Produce messages with keys
print("\nProducing messages:")
messages = [
('user1', {'order_id': 1, 'amount': 100}),
('user2', {'order_id': 2, 'amount': 200}),
('user1', {'order_id': 3, 'amount': 150}),
('user3', {'order_id': 4, 'amount': 300}),
('user2', {'order_id': 5, 'amount': 250}),
('user1', {'order_id': 6, 'amount': 175}),
]
for key, msg in messages:
partition_id, offset = topic.produce(msg, key)
print(f" Key={key}, Order={msg['order_id']} → Partition {partition_id}, Offset {offset}")
# Show partitions
print("\nPartition contents:")
for p in topic.partitions:
print(f" Partition {p.id} ({p.size()} messages):")
for msg in p.log:
print(f" Offset {msg['offset']}: {msg['message']}")
# Show ordering
print("\nKey ordering:")
user1_partition = topic.get_partition('user1')
print(f" user1 always goes to partition {user1_partition}")
user1_msgs = [m for m in topic.partitions[user1_partition].log]
for m in user1_msgs:
print(f" Offset {m['offset']}: {m['message']}")
# TOPIC BEST PRACTICES
print("\n=== Topic Best Practices ===")
print("""
1. NUMBER OF PARTITIONS
- Based on throughput
- More partitions = more parallelism
- But more overhead
- Rule: 1-10 per broker core
2. REPLICATION FACTOR
- 3 for production
- 2 for less critical
- 1 for dev only
3. KEY DESIGN
- Same key → same partition
- Use for ordering
- Balance partitions
4. RETENTION
- Time-based: 7 days default
- Size-based: cap disk
- Compaction: keep latest
5. NAMING
- Descriptive
- Consistent
- Avoid special chars
6. MONITORING
- Partition lag
- Under-replicated partitions
- Disk usage
""")
# TOPIC SUMMARY
print("\n=== Topic & Partition Summary ===")
print(f"{'Concept':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Topic':<25} {'Message category':<40}")
print(f"{'Partition':<25} {'Ordered log':<40}")
print(f"{'Offset':<25} {'Position in partition':<40}")
print(f"{'Ordering':<25} {'Per partition':<40}")
print(f"{'Parallelism':<25} {'One consumer per partition':<40}")
print(f"{'Retention':<25} {'Time or size based':<40}")📤 Producers
🎯 Publishing Messages
Producers publish messages to Kafka topics. They choose partitions, batch messages, and handle acknowledgments for durability.
Producers
Partitioning: Key-based or round-robin. Batching: Group messages. Compression: Reduce size. Acks: Durability level. Retries: Handle failures. Idempotence: Exactly-once.
📤 Partitioning📤 Batching📤 Compression📤 Acks
# Kafka Producers
print("=== Producer Overview ===")
print("""
Producer:
- Publishes messages to topics
- Chooses partition
- Batches messages
- Compresses data
- Handles acknowledgments
- Retries on failure
Flow:
1. Create producer
2. Send message
3. Producer batches
4. Sends to broker
5. Broker acknowledges
6. Producer handles response
""")
print("\n=== Producer Configuration ===")
print("""
Key configurations:
bootstrap.servers:
- Kafka brokers
- 'localhost:9092'
key.serializer:
- Serialize key
- StringSerializer, IntegerSerializer
value.serializer:
- Serialize value
- StringSerializer, JsonSerializer
acks:
- 0: No acknowledgment (fastest)
- 1: Leader acknowledgment
- all: All in-sync replicas (safest)
retries:
- Number of retries
- Default: 2147483647
batch.size:
- Batch size in bytes
- Default: 16384
linger.ms:
- Wait time for batching
- Default: 0
compression.type:
- none, gzip, snappy, lz4, zstd
enable.idempotence:
- Exactly-once semantics
- true/false
""")
print("\n=== Acknowledgment Modes ===")
print("""
acks=0:
- Fire and forget
- Fastest
- May lose messages
acks=1:
- Leader acknowledges
- Balance
- May lose on leader failure
acks=all:
- All ISR acknowledge
- Safest
- Slowest
- No data loss (with min.insync.replicas)
""")
# IMPLEMENTATION
print("\n=== Simulating Producer ===")
class Producer:
"""Simulate Kafka Producer."""
def __init__(self, broker, acks='all', batch_size=16384, linger_ms=0,
compression='none', retries=3, idempotent=False):
self.broker = broker
self.acks = acks
self.batch_size = batch_size
self.linger_ms = linger_ms
self.compression = compression
self.retries = retries
self.idempotent = idempotent
self.batch = []
self.stats = {
'sent': 0,
'failed': 0,
'retried': 0,
'batches': 0,
}
def send(self, topic, key, value):
"""Send a message."""
# Add to batch
self.batch.append({
'topic': topic,
'key': key,
'value': value,
'timestamp': time.time(),
})
# Check if batch is full
if len(self.batch) >= 100: # Simplified batch size
self.flush()
return True
def flush(self):
"""Send batch to broker."""
if not self.batch:
return
print(f" Flushing batch of {len(self.batch)} messages")
for msg in self.batch:
# Simulate send
try:
# In real Kafka, this would send to broker
self.stats['sent'] += 1
except Exception as e:
self.stats['failed'] += 1
self.stats['batches'] += 1
self.batch = []
def get_stats(self):
"""Get producer stats."""
return self.stats
# Demo
print("=== Producer Demo ===")
# Create mock broker
class MockBroker:
def __init__(self):
self.messages = []
def produce(self, topic, key, value):
self.messages.append({'topic': topic, 'key': key, 'value': value})
return len(self.messages) - 1
broker = MockBroker()
# Create producer with different acks
producers = {
'acks=0': Producer(broker, acks=0),
'acks=1': Producer(broker, acks=1),
'acks=all': Producer(broker, acks='all'),
}
print("Sending messages with different acks:")
for name, producer in producers.items():
print(f"\n{name}:")
for i in range(5):
producer.send('orders', f'key{i}', f'value{i}')
producer.flush()
print(f" Stats: {producer.get_stats()}")
# Producer with batching
print("\n=== Producer with Batching ===")
batch_producer = Producer(
broker,
acks='all',
batch_size=16384,
linger_ms=100,
compression='snappy',
)
# Send many messages
for i in range(250):
batch_producer.send('events', f'key{i}', f'event{i}')
batch_producer.flush()
print(f"Stats: {batch_producer.get_stats()}")
# PRODUCER BEST PRACTICES
print("\n=== Producer Best Practices ===")
print("""
1. USE ACKS=ALL
- Durability
- No data loss
2. ENABLE IDEMPOTENCE
- Exactly-once
- Avoid duplicates
3. BATCH AND COMPRESS
- Higher throughput
- Lower network usage
4. TUNE LINGER.MS
- Balance latency vs throughput
- 5-100ms
5. HANDLE ERRORS
- Retries
- Callbacks
- Dead letter queue
6. MONITOR
- Send rate
- Error rate
- Latency
7. KEY DESIGN
- Consistent partition
- Balance load
8. AVOID LARGE MESSAGES
- Max 1 MB default
- Split if needed
""")
# PRODUCER SUMMARY
print("\n=== Producer Summary ===")
print(f"{'Config':<25} {'Values':<30} {'Recommendation':<20}")
print("-" * 75)
print(f"{'acks':<25} {'0, 1, all':<30} {'all':<20}")
print(f"{'retries':<25} {'Integer':<30} {'3+':<20}")
print(f"{'batch.size':<25} {'Bytes':<30} {'16384-1048576':<20}")
print(f"{'linger.ms':<25} {'Milliseconds':<30} {'5-100':<20}")
print(f"{'compression':<25} {'none, gzip, snappy, lz4, zstd':<30} {'snappy/lz4':<20}")
print(f"{'idempotence':<25} {'true/false':<30} {'true':<20}")📥 Consumers
🎯 Reading Messages
Consumers subscribe to topics and read messages. They track offsets, commit progress, and work in consumer groups for parallel consumption.
Consumers
Subscribe: Choose topics. Poll: Fetch messages. Offset: Track position. Commit: Save progress. Group: Parallel consumption. Rebalance: Partition assignment.
📥 Subscribe📥 Poll📥 Offset📥 Group
# Kafka Consumers
print("=== Consumer Overview ===")
print("""
Consumer:
- Subscribes to topics
- Reads messages
- Tracks offset
- Commits progress
- Works in consumer groups
- Handles rebalancing
Flow:
1. Create consumer
2. Subscribe to topic
3. Poll for messages
4. Process messages
5. Commit offset
6. Repeat
""")
print("\n=== Consumer Configuration ===")
print("""
Key configurations:
bootstrap.servers:
- Kafka brokers
group.id:
- Consumer group
- Required for groups
key.deserializer:
- Deserialize key
- StringDeserializer
value.deserializer:
- Deserialize value
- StringDeserializer, JsonDeserializer
auto.offset.reset:
- earliest: From beginning
- latest: From end
- none: Error if no offset
enable.auto.commit:
- true: Auto commit
- false: Manual commit
auto.commit.interval.ms:
- Auto commit interval
- Default: 5000
max.poll.records:
- Max records per poll
- Default: 500
session.timeout.ms:
- Consumer timeout
- Default: 10000
heartbeat.interval.ms:
- Heartbeat interval
- Default: 3000
""")
print("\n=== Offset Management ===")
print("""
Offset:
- Position in partition
- Tracked per consumer group
- Committed to Kafka
Auto commit:
- enable.auto.commit=true
- Every 5 seconds
- May lose messages on failure
Manual commit:
- enable.auto.commit=false
- consumer.commitSync()
- At-least-once processing
Commit strategies:
- At-most-once: Commit before processing
- At-least-once: Commit after processing
- Exactly-once: Transactions
""")
print("\n=== Consumer Groups ===")
print("""
Consumer Group:
- Set of consumers
- Same group.id
- Partitions distributed
- Each partition → one consumer
Example:
Topic: orders (3 partitions)
Group: order-processors
Consumer 1: Partition 0
Consumer 2: Partition 1
Consumer 3: Partition 2
If Consumer 2 fails:
Consumer 1: Partition 0, 1
Consumer 3: Partition 2
""")
# IMPLEMENTATION
print("\n=== Simulating Consumer ===")
class Consumer:
"""Simulate Kafka Consumer."""
def __init__(self, broker, group_id, auto_commit=True,
auto_offset_reset='latest', max_poll_records=500):
self.broker = broker
self.group_id = group_id
self.auto_commit = auto_commit
self.auto_offset_reset = auto_offset_reset
self.max_poll_records = max_poll_records
self.subscriptions = []
self.offsets = defaultdict(int)
self.committed_offsets = defaultdict(int)
self.assigned_partitions = []
self.stats = {
'polled': 0,
'processed': 0,
'committed': 0,
}
def subscribe(self, topics):
"""Subscribe to topics."""
if isinstance(topics, str):
topics = [topics]
self.subscriptions = topics
print(f"Consumer {self.group_id}: Subscribed to {topics}")
def assign(self, partitions):
"""Assign partitions."""
self.assigned_partitions = partitions
print(f"Consumer {self.group_id}: Assigned partitions {partitions}")
def poll(self, timeout_ms=1000):
"""Poll for messages."""
messages = []
for topic in self.subscriptions:
num_partitions = self.broker.get_partition_count(topic)
for partition in range(num_partitions):
if self.assigned_partitions and partition not in self.assigned_partitions:
continue
offset = self.offsets[(topic, partition)]
# Get message
msg = self.broker.consume(topic, partition, offset)
if msg:
messages.append({
'topic': topic,
'partition': partition,
'offset': msg['offset'],
'value': msg['message'],
})
self.offsets[(topic, partition)] = offset + 1
self.stats['polled'] += 1
if len(messages) >= self.max_poll_records:
break
return messages
def commit(self):
"""Commit offsets."""
if not self.auto_commit:
for key, offset in self.offsets.items():
self.committed_offsets[key] = offset
self.stats['committed'] += 1
print(f"Consumer {self.group_id}: Committed offsets")
def close(self):
"""Close consumer."""
if self.auto_commit:
self.commit()
print(f"Consumer {self.group_id}: Closed")
def get_stats(self):
return self.stats
# Demo
print("=== Consumer Demo ===")
# Reuse broker from earlier (in practice, create new)
class SimpleBroker:
def __init__(self):
self.topics = {}
def create_topic(self, name, partitions):
self.topics[name] = {i: [] for i in range(partitions)}
def produce(self, topic, partition, message):
self.topics[topic][partition].append({
'offset': len(self.topics[topic][partition]),
'message': message,
})
def consume(self, topic, partition, offset):
if topic in self.topics and partition in self.topics[topic]:
msgs = self.topics[topic][partition]
if offset < len(msgs):
return msgs[offset]
return None
def get_partition_count(self, topic):
return len(self.topics.get(topic, {}))
broker = SimpleBroker()
broker.create_topic('orders', 3)
# Produce messages
for i in range(10):
partition = i % 3
broker.produce('orders', partition, f'message-{i}')
# Create consumer
consumer = Consumer(broker, 'order-processor', auto_commit=False)
consumer.subscribe('orders')
# Poll messages
print("\nPolling messages:")
for _ in range(15):
messages = consumer.poll()
if not messages:
break
for msg in messages:
print(f" {msg['topic']}[{msg['partition']}] offset={msg['offset']}: {msg['value']}")
consumer.stats['processed'] += 1
# Commit
consumer.commit()
consumer.close()
print(f"\nStats: {consumer.get_stats()}")
# CONSUMER GROUP REBALANCE
print("\n=== Consumer Group Rebalance ===")
class ConsumerGroup:
"""Simulate Kafka Consumer Group."""
def __init__(self, group_id, topic, num_partitions):
self.group_id = group_id
self.topic = topic
self.num_partitions = num_partitions
self.consumers = []
self.assignments = {}
def add_consumer(self, consumer_id):
"""Add consumer to group."""
self.consumers.append(consumer_id)
self.rebalance()
def remove_consumer(self, consumer_id):
"""Remove consumer from group."""
if consumer_id in self.consumers:
self.consumers.remove(consumer_id)
self.rebalance()
def rebalance(self):
"""Rebalance partitions."""
self.assignments = {}
num_consumers = len(self.consumers)
if num_consumers == 0:
return
for i, partition in enumerate(range(self.num_partitions)):
consumer = self.consumers[i % num_consumers]
if consumer not in self.assignments:
self.assignments[consumer] = []
self.assignments[consumer].append(partition)
print(f" Rebalanced: {self.assignments}")
def get_assignments(self):
return self.assignments
# Demo
print("Consumer Group Rebalance Demo:")
group = ConsumerGroup('processors', 'orders', 6)
print("\n1 consumer:")
group.add_consumer('consumer-1')
print("\n2 consumers:")
group.add_consumer('consumer-2')
print("\n3 consumers:")
group.add_consumer('consumer-3')
print("\nConsumer 2 fails:")
group.remove_consumer('consumer-2')
# CONSUMER BEST PRACTICES
print("\n=== Consumer Best Practices ===")
print("""
1. USE CONSUMER GROUPS
- Parallel consumption
- Fault tolerance
2. COMMIT OFFSETS MANUALLY
- Control processing
- At-least-once
3. HANDLE REBALANCING
- Commit before rebalance
- Cleanup resources
4. TUNE MAX.POLL.RECORDS
- Balance batch size
- Avoid timeout
5. MONITOR LAG
- Consumer lag
- Alert on high lag
6. IDEMPOTENT PROCESSING
- Handle duplicates
- Exactly-once
7. ERROR HANDLING
- Dead letter queue
- Retry logic
8. SESSION TIMEOUT
- Balance detection
- Avoid false positives
""")
# CONSUMER SUMMARY
print("\n=== Consumer Summary ===")
print(f"{'Config':<30} {'Values':<25} {'Recommendation':<20}")
print("-" * 75)
print(f"{'group.id':<30} {'String':<25} {'Unique name':<20}")
print(f"{'auto.offset.reset':<30} {'earliest, latest':<25} {'earliest':<20}")
print(f"{'enable.auto.commit':<30} {'true/false':<25} {'false':<20}")
print(f"{'max.poll.records':<30} {'Integer':<25} {'500':<20}")
print(f"{'session.timeout.ms':<30} {'Milliseconds':<25} {'10000':<20}")🖥️ Brokers & Clusters
🎯 Kafka Servers
Brokers are Kafka servers that store messages and serve clients. A cluster is a group of brokers working together for scalability and fault tolerance.
Brokers & Clusters
Broker: Kafka server. Cluster: Multiple brokers. Controller: Leader broker. Leader: Partition leader. Follower: Replica. ISR: In-sync replicas.
🖥️ Broker🖥️ Cluster🖥️ Controller🖥️ ISR
# Kafka Brokers and Clusters
print("=== What is a Broker? ===")
print("""
Broker:
- Kafka server
- Stores messages
- Serves producers/consumers
- Replicates data
- Handles partitions
Each broker:
- Has unique ID
- Stores partitions
- Can be leader or follower
- Communicates with other brokers
""")
print("\n=== Kafka Cluster ===")
print("""
Cluster:
- Multiple brokers
- Distributed storage
- Fault tolerance
- Scalability
Example:
Broker 1: Topic A P0 (leader), Topic B P1 (follower)
Broker 2: Topic A P1 (leader), Topic B P0 (follower)
Broker 3: Topic A P2 (leader), Topic B P2 (follower)
Benefits:
- Load distribution
- Fault tolerance
- Scalability
""")
print("\n=== Controller ===")
print("""
Controller:
- One broker in cluster
- Elected by ZooKeeper/KRaft
- Manages partitions
- Leader election
- Replica assignment
Responsibilities:
- Partition leadership
- Broker failures
- Topic creation/deletion
- Replica management
""")
print("\n=== In-Sync Replicas (ISR) ===")
print("""
ISR:
- Replicas caught up with leader
- Eligible for leader election
- Tracked by controller
Example:
Partition 0:
Leader: Broker 1
Replicas: Broker 2, Broker 3
ISR: [1, 2, 3]
If Broker 3 lags:
ISR: [1, 2]
If Broker 3 catches up:
ISR: [1, 2, 3]
Configuration:
min.insync.replicas=2
(At least 2 replicas must acknowledge)
""")
# IMPLEMENTATION
print("\n=== Simulating Brokers ===")
class Broker:
"""Simulate Kafka Broker."""
def __init__(self, broker_id, host='localhost', port=9092):
self.id = broker_id
self.host = host
self.port = port
self.partitions = {} # topic -> {partition -> data}
self.is_controller = False
self.replicas = {} # topic -> {partition -> [broker_ids]}
self.isr = {} # topic -> {partition -> [broker_ids]}
def create_partition(self, topic, partition, is_leader=False):
"""Create a partition."""
if topic not in self.partitions:
self.partitions[topic] = {}
self.partitions[topic][partition] = {
'leader': is_leader,
'log': [],
'offset': 0,
}
if is_leader:
print(f"Broker {self.id}: Leader for {topic}-{partition}")
else:
print(f"Broker {self.id}: Follower for {topic}-{partition}")
def produce(self, topic, partition, message):
"""Produce to partition."""
if topic in self.partitions and partition in self.partitions[topic]:
p = self.partitions[topic][partition]
if p['leader']:
offset = p['offset']
p['log'].append({'offset': offset, 'message': message})
p['offset'] += 1
return offset
return None
def consume(self, topic, partition, offset):
"""Consume from partition."""
if topic in self.partitions and partition in self.partitions[topic]:
p = self.partitions[topic][partition]
for msg in p['log']:
if msg['offset'] == offset:
return msg
return None
def get_info(self):
"""Get broker info."""
return {
'id': self.id,
'host': self.host,
'port': self.port,
'controller': self.is_controller,
'partitions': sum(len(p) for p in self.partitions.values()),
}
class KafkaCluster:
"""Simulate Kafka Cluster."""
def __init__(self, cluster_id):
self.cluster_id = cluster_id
self.brokers = {}
self.controller_id = None
self.topics = {} # topic -> {partitions, replication}
def add_broker(self, broker):
"""Add broker to cluster."""
self.brokers[broker.id] = broker
print(f"Cluster: Added broker {broker.id}")
# First broker becomes controller
if self.controller_id is None:
self.controller_id = broker.id
broker.is_controller = True
print(f"Cluster: Broker {broker.id} is controller")
def remove_broker(self, broker_id):
"""Remove broker."""
if broker_id in self.brokers:
del self.brokers[broker_id]
print(f"Cluster: Removed broker {broker_id}")
# Elect new controller
if broker_id == self.controller_id and self.brokers:
new_controller = list(self.brokers.keys())[0]
self.controller_id = new_controller
self.brokers[new_controller].is_controller = True
print(f"Cluster: Broker {new_controller} is new controller")
def create_topic(self, topic, num_partitions, replication_factor):
"""Create topic across brokers."""
broker_ids = list(self.brokers.keys())
if len(broker_ids) < replication_factor:
print(f"Cluster: Not enough brokers for replication {replication_factor}")
return
self.topics[topic] = {
'partitions': num_partitions,
'replication': replication_factor,
}
print(f"\nCluster: Creating topic '{topic}'")
print(f" Partitions: {num_partitions}, Replication: {replication_factor}")
# Assign partitions to brokers
for partition in range(num_partitions):
# Leader = round-robin
leader_id = broker_ids[partition % len(broker_ids)]
# Create partition on leader
self.brokers[leader_id].create_partition(topic, partition, is_leader=True)
# Create replicas on followers
for i in range(1, replication_factor):
follower_id = broker_ids[(partition + i) % len(broker_ids)]
self.brokers[follower_id].create_partition(topic, partition, is_leader=False)
# Track ISR
isr = [leader_id]
for i in range(1, replication_factor):
follower_id = broker_ids[(partition + i) % len(broker_ids)]
isr.append(follower_id)
# Store ISR
for bid in isr:
if topic not in self.brokers[bid].isr:
self.brokers[bid].isr[topic] = {}
self.brokers[bid].isr[topic][partition] = isr
def get_status(self):
"""Get cluster status."""
return {
'cluster_id': self.cluster_id,
'brokers': len(self.brokers),
'controller': self.controller_id,
'topics': len(self.topics),
'broker_info': [b.get_info() for b in self.brokers.values()],
}
# Demo
print("=== Cluster Demo ===")
# Create cluster
cluster = KafkaCluster('kafka-cluster-1')
# Add brokers
for i in range(1, 4):
broker = Broker(i, f'broker{i}', 9092 + i)
cluster.add_broker(broker)
# Create topic
cluster.create_topic('orders', num_partitions=3, replication_factor=2)
# Show cluster status
print("\n=== Cluster Status ===")
status = cluster.get_status()
print(f"Cluster: {status['cluster_id']}")
print(f"Brokers: {status['brokers']}")
print(f"Controller: {status['controller']}")
print(f"Topics: {status['topics']}")
for info in status['broker_info']:
print(f" Broker {info['id']}: {info['partitions']} partitions, "
f"controller={info['controller']}")
# Produce messages
print("\n=== Producing Messages ===")
broker1 = cluster.brokers[1]
for i in range(5):
offset = broker1.produce('orders', 0, f'message-{i}')
if offset is not None:
print(f" Broker 1: Produced offset {offset}")
# Simulate broker failure
print("\n=== Simulating Broker Failure ===")
cluster.remove_broker(2)
# Show status
print("\n=== Status After Failure ===")
status = cluster.get_status()
print(f"Brokers: {status['brokers']}")
print(f"Controller: {status['controller']}")
# BROKER BEST PRACTICES
print("\n=== Broker Best Practices ===")
print("""
1. USE MULTIPLE BROKERS
- Minimum 3 for production
- Fault tolerance
- Load distribution
2. REPLICATION FACTOR 3
- 2 brokers can fail
- Data durability
- High availability
3. MONITOR BROKERS
- CPU, memory, disk
- Network
- Partition count
4. BALANCE PARTITIONS
- Even distribution
- Avoid hotspots
- Use kafka-reassign-partitions
5. CONFIGURE LOG RETENTION
- Time-based
- Size-based
- Compaction
6. USE DEDICATED DISKS
- Separate from OS
- SSD for performance
- RAID for redundancy
7. TUNE JVM
- Heap size
- GC settings
- Direct memory
8. MONITOR ISR
- Under-replicated partitions
- ISR shrinks
- Alert on issues
""")
# BROKER SUMMARY
print("\n=== Broker Summary ===")
print(f"{'Component':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Broker':<25} {'Kafka server':<40}")
print(f"{'Cluster':<25} {'Multiple brokers':<40}")
print(f"{'Controller':<25} {'Leader broker':<40}")
print(f"{'Leader':<25} {'Partition leader':<40}")
print(f"{'Follower':<25} {'Replica':<40}")
print(f"{'ISR':<25} {'In-sync replicas':<40}")🔄 Replication
🎯 Fault Tolerance
Replication copies each partition to multiple brokers. It ensures data durability and availability even if brokers fail.
Replication
Replication factor: Number of copies. Leader: Handles reads/writes. Follower: Replicates leader. ISR: In-sync replicas. Acks: Durability. Failover: New leader.
🔄 Replication🔄 Leader🔄 Follower🔄 ISR
# Kafka Replication
print("=== What is Replication? ===")
print("""
Replication:
- Copies of each partition
- Stored on different brokers
- Fault tolerance
- High availability
Replication factor:
- 1: No fault tolerance
- 2: 1 broker can fail
- 3: 2 brokers can fail (recommended)
""")
print("\n=== Leader and Followers ===")
print("""
For each partition:
- One leader
- Multiple followers
- Leader handles reads/writes
- Followers replicate
Example (replication=3):
Partition 0:
Leader: Broker 1
Follower: Broker 2
Follower: Broker 3
Producers → Leader (Broker 1)
Consumers → Leader (Broker 1)
Followers replicate from leader
""")
print("\n=== ISR (In-Sync Replicas) ===")
print("""
ISR:
- Replicas caught up with leader
- Eligible for leader election
- Tracked by controller
Example:
Partition 0:
Leader: Broker 1 (offset 100)
Follower: Broker 2 (offset 100) ← in ISR
Follower: Broker 3 (offset 95) ← out of ISR
ISR: [1, 2]
If Broker 1 fails:
New leader: Broker 2 (from ISR)
Configuration:
min.insync.replicas=2
(At least 2 replicas must acknowledge)
""")
print("\n=== Acknowledgment and Durability ===")
print("""
acks=0:
- No acknowledgment
- May lose messages
acks=1:
- Leader acknowledges
- May lose on leader failure
acks=all:
- All ISR acknowledge
- No data loss
- Requires min.insync.replicas
Durability matrix:
acks | min.insync | Durability
------|------------|----------
0 | any | None
1 | any | Low
all | 1 | Medium
all | 2 | High
all | 3 | Highest
""")
# IMPLEMENTATION
print("\n=== Simulating Replication ===")
class ReplicatedPartition:
"""Simulate replicated partition."""
def __init__(self, topic, partition_id, replication_factor=3):
self.topic = topic
self.partition_id = partition_id
self.replication_factor = replication_factor
# Replicas: broker_id -> {log, offset}
self.replicas = {}
self.leader_id = None
self.isr = []
def add_replica(self, broker_id, is_leader=False):
"""Add a replica."""
self.replicas[broker_id] = {
'log': [],
'offset': 0,
'is_leader': is_leader,
}
if is_leader:
self.leader_id = broker_id
self.isr.append(broker_id)
print(f" Partition {self.partition_id}: Added replica on broker {broker_id}"
f"{' (leader)' if is_leader else ''}")
def produce(self, message, acks='all'):
"""Produce a message."""
if self.leader_id is None:
return None
# Write to leader
leader = self.replicas[self.leader_id]
offset = leader['offset']
leader['log'].append({'offset': offset, 'message': message})
leader['offset'] += 1
# Replicate to followers
replicated = 0
for broker_id, replica in self.replicas.items():
if broker_id == self.leader_id:
continue
# Simulate replication
if broker_id in self.isr:
replica['log'].append({'offset': offset, 'message': message})
replica['offset'] += 1
replicated += 1
# Check acks
if acks == 'all':
if replicated + 1 < len(self.isr):
print(f" WARNING: Not enough replicas acknowledged")
return None
return offset
def fail_broker(self, broker_id):
"""Simulate broker failure."""
if broker_id not in self.replicas:
return
print(f" Partition {self.partition_id}: Broker {broker_id} failed")
# Remove from ISR
if broker_id in self.isr:
self.isr.remove(broker_id)
# If leader failed, elect new leader
if broker_id == self.leader_id:
if self.isr:
new_leader = self.isr[0]
self.leader_id = new_leader
self.replicas[new_leader]['is_leader'] = True
print(f" Partition {self.partition_id}: New leader is broker {new_leader}")
else:
self.leader_id = None
print(f" Partition {self.partition_id}: No leader available!")
def get_status(self):
"""Get partition status."""
return {
'topic': self.topic,
'partition': self.partition_id,
'leader': self.leader_id,
'isr': self.isr,
'replicas': list(self.replicas.keys()),
'offsets': {bid: r['offset'] for bid, r in self.replicas.items()},
}
class ReplicatedTopic:
"""Simulate replicated topic."""
def __init__(self, name, num_partitions=3, replication_factor=3):
self.name = name
self.num_partitions = num_partitions
self.replication_factor = replication_factor
self.partitions = []
# Create partitions
for i in range(num_partitions):
partition = ReplicatedPartition(name, i, replication_factor)
self.partitions.append(partition)
# Assign replicas (simplified)
broker_ids = list(range(1, 4)) # 3 brokers
for i, partition in enumerate(self.partitions):
# Leader = round-robin
leader_id = broker_ids[i % len(broker_ids)]
partition.add_replica(leader_id, is_leader=True)
# Followers
for j in range(1, replication_factor):
follower_id = broker_ids[(i + j) % len(broker_ids)]
partition.add_replica(follower_id, is_leader=False)
def produce(self, partition_id, message, acks='all'):
"""Produce to a partition."""
if 0 <= partition_id < self.num_partitions:
return self.partitions[partition_id].produce(message, acks)
return None
def fail_broker(self, broker_id):
"""Fail a broker."""
print(f"\n=== Broker {broker_id} Failure ===")
for partition in self.partitions:
partition.fail_broker(broker_id)
def get_status(self):
"""Get topic status."""
return {
'name': self.name,
'partitions': self.num_partitions,
'replication': self.replication_factor,
'partition_status': [p.get_status() for p in self.partitions],
}
# Demo
print("=== Replication Demo ===")
# Create replicated topic
topic = ReplicatedTopic('orders', num_partitions=3, replication_factor=3)
print("\n=== Topic Created ===")
status = topic.get_status()
print(f"Topic: {status['name']}")
print(f"Partitions: {status['partitions']}")
print(f"Replication: {status['replication']}")
# Produce messages
print("\n=== Producing Messages ===")
for i in range(6):
partition_id = i % 3
offset = topic.produce(partition_id, f'message-{i}')
print(f" Partition {partition_id}, message {i} → offset {offset}")
# Show status
print("\n=== Partition Status ===")
for p_status in topic.get_status()['partition_status']:
print(f" Partition {p_status['partition']}:")
print(f" Leader: {p_status['leader']}")
print(f" ISR: {p_status['isr']}")
print(f" Offsets: {p_status['offsets']}")
# Fail broker
topic.fail_broker(2)
# Show status after failure
print("\n=== After Failure ===")
for p_status in topic.get_status()['partition_status']:
print(f" Partition {p_status['partition']}:")
print(f" Leader: {p_status['leader']}")
print(f" ISR: {p_status['isr']}")
print(f" Offsets: {p_status['offsets']}")
# Produce more messages
print("\n=== Producing After Failure ===")
for i in range(3):
partition_id = i % 3
offset = topic.produce(partition_id, f'after-failure-{i}')
print(f" Partition {partition_id}, message {i} → offset {offset}")
# REPLICATION SUMMARY
print("\n=== Replication Summary ===")
print(f"{'Aspect':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Replication factor':<25} {'Number of copies':<40}")
print(f"{'Leader':<25} {'Handles reads/writes':<40}")
print(f"{'Follower':<25} {'Replicates leader':<40}")
print(f"{'ISR':<25} {'In-sync replicas':<40}")
print(f"{'acks=all':<25} {'All ISR acknowledge':<40}")
print(f"{'min.insync.replicas':<25} {'Min replicas for write':<40}")
print(f"{'Failover':<25} {'New leader from ISR':<40}")📍 Offsets & Consumer Groups
🎯 Tracking Progress
Offsets are positions in partitions. Consumer groups coordinate consumers, with each partition assigned to one consumer for parallel processing.
Offsets & Groups
Offset: Position in partition. Commit: Save progress. Group: Set of consumers. Assignment: Partition to consumer. Rebalance: Redistribution. Lag: Unprocessed messages.
📍 Offset📍 Commit📍 Group📍 Rebalance
# Offsets and Consumer Groups
print("=== What is an Offset? ===")
print("""
Offset:
- Position in a partition
- Unique per partition
- Incremented per message
- Used by consumers to track progress
Example:
Partition 0:
Offset 0: message A
Offset 1: message B
Offset 2: message C
Offset 3: message D
Consumer at offset 2 → next read is message D
""")
print("\n=== Offset Management ===")
print("""
Consumer offset:
- Last processed message
- Stored in Kafka (__consumer_offsets topic)
- Per consumer group per partition
Auto commit:
enable.auto.commit=true
auto.commit.interval.ms=5000
- Commits every 5 seconds
- May lose messages on failure
Manual commit:
enable.auto.commit=false
consumer.commitSync()
- After processing
- At-least-once
Commit strategies:
- At-most-once: Commit before processing
- At-least-once: Commit after processing
- Exactly-once: Transactions
""")
print("\n=== Consumer Groups ===")
print("""
Consumer Group:
- Set of consumers
- Same group.id
- Partitions distributed
- Each partition → one consumer
Example:
Topic: orders (3 partitions)
Group: order-processors (2 consumers)
Consumer 1: Partition 0, Partition 1
Consumer 2: Partition 2
If Consumer 2 fails:
Consumer 1: Partition 0, 1, 2
Maximum parallelism:
num_consumers = num_partitions
More consumers = idle
""")
print("\n=== Rebalancing ===")
print("""
Rebalance:
- Partition reassignment
- Triggered by:
- Consumer joins
- Consumer leaves
- Partition changes
Types:
- Eager: Stop all, reassign (default)
- Cooperative: Incremental (newer)
During rebalance:
- Consumers stop
- Partitions reassigned
- Consumers resume
Best practices:
- Commit before rebalance
- Handle rebalance listener
- Avoid frequent rebalances
""")
# IMPLEMENTATION
print("\n=== Simulating Offsets & Groups ===")
class OffsetManager:
"""Simulate Kafka offset management."""
def __init__(self):
# group_id -> topic -> partition -> offset
self.offsets = {}
self.commits = []
def commit(self, group_id, topic, partition, offset):
"""Commit an offset."""
if group_id not in self.offsets:
self.offsets[group_id] = {}
if topic not in self.offsets[group_id]:
self.offsets[group_id][topic] = {}
self.offsets[group_id][topic][partition] = offset
self.commits.append({
'group': group_id,
'topic': topic,
'partition': partition,
'offset': offset,
'timestamp': time.time(),
})
def get_offset(self, group_id, topic, partition):
"""Get committed offset."""
return self.offsets.get(group_id, {}).get(topic, {}).get(partition, -1)
def get_group_offsets(self, group_id):
"""Get all offsets for group."""
return self.offsets.get(group_id, {})
class ConsumerGroupManager:
"""Simulate consumer group management."""
def __init__(self, group_id):
self.group_id = group_id
self.consumers = []
self.assignments = {} # consumer_id -> [partitions]
self.topic_partitions = {}
def set_topic(self, topic, num_partitions):
"""Set topic partitions."""
self.topic_partitions[topic] = num_partitions
def add_consumer(self, consumer_id):
"""Add consumer."""
if consumer_id not in self.consumers:
self.consumers.append(consumer_id)
self.rebalance()
def remove_consumer(self, consumer_id):
"""Remove consumer."""
if consumer_id in self.consumers:
self.consumers.remove(consumer_id)
self.rebalance()
def rebalance(self):
"""Rebalance partitions."""
self.assignments = {}
if not self.consumers:
return
# Get all partitions
all_partitions = []
for topic, num in self.topic_partitions.items():
for p in range(num):
all_partitions.append((topic, p))
# Distribute round-robin
for i, partition in enumerate(all_partitions):
consumer = self.consumers[i % len(self.consumers)]
if consumer not in self.assignments:
self.assignments[consumer] = []
self.assignments[consumer].append(partition)
print(f" Rebalanced: {self.assignments}")
def get_lag(self, consumer_id, committed_offsets, latest_offsets):
"""Calculate consumer lag."""
lag = 0
for topic, partition in self.assignments.get(consumer_id, []):
committed = committed_offsets.get(topic, {}).get(partition, 0)
latest = latest_offsets.get(topic, {}).get(partition, 0)
lag += max(0, latest - committed)
return lag
# Demo
print("=== Offset & Group Demo ===")
# Offset manager
offset_mgr = OffsetManager()
# Consumer group
group = ConsumerGroupManager('order-processors')
group.set_topic('orders', 3)
print("\n1. One consumer:")
group.add_consumer('consumer-1')
print("\n2. Two consumers:")
group.add_consumer('consumer-2')
print("\n3. Three consumers:")
group.add_consumer('consumer-3')
print("\n4. Four consumers (one idle):")
group.add_consumer('consumer-4')
print("\n5. Consumer 2 fails:")
group.remove_consumer('consumer-2')
# Offset commit demo
print("\n=== Offset Commit Demo ===")
# Simulate processing
for i in range(10):
partition = i % 3
offset_mgr.commit('order-processors', 'orders', partition, i)
print("Committed offsets:")
for topic, partitions in offset_mgr.get_group_offsets('order-processors').items():
for partition, offset in partitions.items():
print(f" {topic}[{partition}]: offset {offset}")
# Lag calculation
latest_offsets = {
'orders': {0: 5, 1: 4, 2: 3}
}
for consumer_id in group.consumers:
lag = group.get_lag(consumer_id,
offset_mgr.get_group_offsets('order-processors'),
latest_offsets)
print(f" {consumer_id}: lag={lag}")
# OFFSET BEST PRACTICES
print("\n=== Offset Best Practices ===")
print("""
1. COMMIT AFTER PROCESSING
- At-least-once
- Avoid data loss
2. USE MANUAL COMMIT
- Control timing
- Handle errors
3. MONITOR LAG
- Consumer lag
- Alert on high lag
4. HANDLE REBALANCE
- Commit before
- Cleanup resources
5. AVOID FREQUENT REBALANCES
- Stable consumers
- Tune timeouts
6. USE COOPERATIVE REBALANCE
- Incremental
- Less disruption
7. STORE OFFSETS
- Kafka (default)
- External (for exactly-once)
8. RESET OFFSETS
- Earliest
- Latest
- Specific offset
""")
# OFFSET SUMMARY
print("\n=== Offset & Group Summary ===")
print(f"{'Concept':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Offset':<25} {'Position in partition':<40}")
print(f"{'Commit':<25} {'Save progress':<40}")
print(f"{'Consumer Group':<25} {'Set of consumers':<40}")
print(f"{'Assignment':<25} {'Partition to consumer':<40}")
print(f"{'Rebalance':<25} {'Redistribution':<40}")
print(f"{'Lag':<25} {'Unprocessed messages':<40}")🔌 Kafka Connect
🎯 Integration Framework
Kafka Connect is a tool for streaming data between Kafka and other systems. It provides connectors for databases, file systems, and cloud services.
Kafka Connect
Source: External → Kafka. Sink: Kafka → External. Connectors: Pre-built. Workers: Run connectors. Transform: Single message. Scale: Distributed.
🔌 Source🔌 Sink🔌 Connectors🔌 Workers
# Kafka Connect
print("=== What is Kafka Connect? ===")
print("""
Kafka Connect:
- Integration framework
- Stream data in/out of Kafka
- Pre-built connectors
- Scalable, fault-tolerant
- No code required
Use cases:
- Database → Kafka
- Kafka → Database
- File → Kafka
- Kafka → S3
- Kafka → Elasticsearch
""")
print("\n=== Source vs Sink ===")
print("""
SOURCE CONNECTOR:
- External system → Kafka
- Examples:
- JDBC Source (database)
- FileStream Source (files)
- S3 Source (S3)
- Debezium (CDC)
SINK CONNECTOR:
- Kafka → External system
- Examples:
- JDBC Sink (database)
- HDFS Sink (HDFS)
- S3 Sink (S3)
- Elasticsearch Sink
Both:
- Configured via JSON
- Run on workers
- Scale horizontally
""")
print("\n=== Connector Configuration ===")
print("""
Source connector (JDBC):
{
"name": "jdbc-source",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:postgresql://localhost:5432/mydb",
"connection.user": "user",
"connection.password": "pass",
"table.whitelist": "users,orders",
"mode": "incrementing",
"incrementing.column.name": "id",
"topic.prefix": "db-"
}
}
Sink connector (HDFS):
{
"name": "hdfs-sink",
"config": {
"connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
"topics": "orders",
"hdfs.url": "hdfs://namenode:8020",
"flush.size": "1000",
"format.class": "io.confluent.connect.hdfs.parquet.ParquetFormat"
}
}
""")
print("\n=== Single Message Transforms (SMT) ===")
print("""
SMT:
- Transform messages in Connect
- No code required
- Chain multiple transforms
Examples:
- MaskField: Mask sensitive data
- ReplaceField: Rename fields
- Filter: Drop messages
- InsertField: Add timestamp
- ExtractField: Extract nested field
Configuration:
"transforms": "mask,insert",
"transforms.mask.type": "org.apache.kafka.connect.transforms.MaskField$Value",
"transforms.mask.fields": "credit_card",
"transforms.insert.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.insert.timestamp.field": "timestamp"
""")
# IMPLEMENTATION
print("\n=== Simulating Kafka Connect ===")
class KafkaConnect:
"""Simulate Kafka Connect."""
def __init__(self):
self.connectors = {}
self.transforms = []
def add_connector(self, name, connector_class, config):
"""Add a connector."""
self.connectors[name] = {
'name': name,
'class': connector_class,
'config': config,
'status': 'RUNNING',
'records_processed': 0,
}
print(f"Connect: Added connector '{name}' ({connector_class})")
def add_transform(self, transform_type, config):
"""Add a transform."""
self.transforms.append({
'type': transform_type,
'config': config,
})
print(f"Connect: Added transform '{transform_type}'")
def process(self, records):
"""Process records through transforms."""
for record in records:
for transform in self.transforms:
record = self._apply_transform(record, transform)
return records
def _apply_transform(self, record, transform):
"""Apply a transform."""
t_type = transform['type']
config = transform['config']
if t_type == 'MaskField':
field = config.get('field')
if field in record:
record[field] = '****'
elif t_type == 'InsertField':
field = config.get('field')
record[field] = config.get('value', 'inserted')
elif t_type == 'ReplaceField':
old = config.get('old')
new = config.get('new')
if old in record:
record[new] = record.pop(old)
return record
def get_status(self):
"""Get connector status."""
return self.connectors
# Demo
print("=== Kafka Connect Demo ===")
connect = KafkaConnect()
# Add source connector
connect.add_connector(
'jdbc-source',
'io.confluent.connect.jdbc.JdbcSourceConnector',
{
'connection.url': 'jdbc:postgresql://localhost:5432/mydb',
'table.whitelist': 'users',
'topic.prefix': 'db-',
}
)
# Add sink connector
connect.add_connector(
'hdfs-sink',
'io.confluent.connect.hdfs.HdfsSinkConnector',
{
'topics': 'orders',
'hdfs.url': 'hdfs://namenode:8020',
}
)
# Add transforms
connect.add_transform('MaskField', {'field': 'credit_card'})
connect.add_transform('InsertField', {'field': 'processed_at', 'value': '2024-01-01'})
# Process records
records = [
{'id': 1, 'name': 'Alice', 'credit_card': '1234-5678-9012-3456'},
{'id': 2, 'name': 'Bob', 'credit_card': '9876-5432-1098-7654'},
]
print("\nOriginal records:")
for r in records:
print(f" {r}")
processed = connect.process(records.copy())
print("\nProcessed records:")
for r in processed:
print(f" {r}")
# Status
print("\nConnector status:")
for name, connector in connect.get_status().items():
print(f" {name}: {connector['status']} ({connector['class']})")
# CONNECT BEST PRACTICES
print("\n=== Kafka Connect Best Practices ===")
print("""
1. USE DISTRIBUTED MODE
- Scalable
- Fault-tolerant
- Production
2. MONITOR CONNECTORS
- Status
- Lag
- Errors
3. USE SMTs
- Lightweight transforms
- No code
4. CONFIGURE OFFSETS
- Auto
- Manual
5. HANDLE ERRORS
- Dead letter queue
- Retry
6. TUNE WORKERS
- Tasks per worker
- Heap size
7. SECURITY
- SSL
- SASL
- ACLs
8. VERSION CONTROL
- Connector configs
- Deploy as code
""")
# CONNECT SUMMARY
print("\n=== Kafka Connect Summary ===")
print(f"{'Component':<25} {'Description':<40}")
print("-" * 65)
print(f"{'Source Connector':<25} {'External → Kafka':<40}")
print(f"{'Sink Connector':<25} {'Kafka → External':<40}")
print(f"{'Worker':<25} {'Runs connectors':<40}")
print(f"{'Transform':<25} {'Message transformation':<40}")
print(f"{'Distributed':<25} {'Scalable, fault-tolerant':<40}")🌊 Kafka Streams
🎯 Stream Processing Library
Kafka Streams is a client library for building stream processing applications. It provides a high-level DSL for transformations, aggregations, and joins.
Kafka Streams
Library: Not a framework. DSL: High-level API. KStream: Stream. KTable: Table. State: Local state stores. Exactly-once: Transactions.
🌊 KStream🌊 KTable🌊 State Store🌊 Topology
# Kafka Streams
print("=== What is Kafka Streams? ===")
print("""
Kafka Streams:
- Client library
- Stream processing
- Part of Kafka
- No separate cluster
- Java/Scala (Kotlin)
Features:
- High-level DSL
- Stateful operations
- Exactly-once
- Fault-tolerant
- Scalable
Concepts:
- KStream: Stream of events
- KTable: Table of latest values
- GlobalKTable: Replicated table
- Topology: Processing graph
""")
print("\n=== KStream vs KTable ===")
print("""
KStream:
- Insert-only
- Every record is new
- Like a log
- Example: clicks, events
KTable:
- Upsert
- Latest value per key
- Like a database table
- Example: user profiles, counts
Example:
KStream: (user1, click), (user2, click), (user1, click)
→ 3 events
KTable: (user1, click), (user2, click), (user1, click)
→ 2 rows (user1, user2)
""")
print("\n=== Kafka Streams DSL ===")
print("""
Operations:
Stateless:
- map, filter, flatMap
- branch, merge
- selectKey
Stateful:
- groupBy, count, reduce
- aggregate
- join, window
Example:
KStream stream = builder.stream("input");
stream
.filter((key, value) -> value.contains("important"))
.mapValues(value -> value.toUpperCase())
.groupByKey()
.count()
.toStream()
.to("output");
""")
print("\n=== State Stores ===")
print("""
State Store:
- Local storage
- RocksDB (default)
- In-memory
- Fault-tolerant (changelog)
Types:
- KeyValueStore
- WindowStore
- SessionStore
Changelog:
- Backed by Kafka topic
- Restore on failure
- Exactly-once
""")
# IMPLEMENTATION
print("\n=== Simulating Kafka Streams ===")
class KStream:
"""Simulate Kafka Streams KStream."""
def __init__(self, records=None):
self.records = records or []
self.operations = []
def filter(self, predicate):
"""Filter records."""
self.operations.append(f"filter({predicate})")
self.records = [r for r in self.records if predicate(r[0], r[1])]
return self
def map(self, mapper):
"""Map records."""
self.operations.append(f"map({mapper})")
self.records = [mapper(k, v) for k, v in self.records]
return self
def mapValues(self, mapper):
"""Map values."""
self.operations.append(f"mapValues({mapper})")
self.records = [(k, mapper(v)) for k, v in self.records]
return self
def groupByKey(self):
"""Group by key."""
self.operations.append("groupByKey()")
return KGroupedStream(self.records)
def to(self, topic):
"""Output to topic."""
self.operations.append(f"to({topic})")
return self.records
class KGroupedStream:
"""Simulate grouped stream."""
def __init__(self, records):
self.records = records
def count(self):
"""Count per key."""
from collections import Counter
counts = Counter(k for k, v in self.records)
return KTable(dict(counts))
def reduce(self, reducer):
"""Reduce values."""
from collections import defaultdict
grouped = defaultdict(list)
for k, v in self.records:
grouped[k].append(v)
result = {}
for k, values in grouped.items():
acc = values[0]
for v in values[1:]:
acc = reducer(acc, v)
result[k] = acc
return KTable(result)
class KTable:
"""Simulate Kafka Streams KTable."""
def __init__(self, data):
self.data = data
def toStream(self):
"""Convert to stream."""
return KStream(list(self.data.items()))
def show(self):
"""Show table."""
for k, v in self.data.items():
print(f" {k}: {v}")
# Demo
print("=== Kafka Streams Demo ===")
# Input records
records = [
('user1', 'click'),
('user2', 'view'),
('user1', 'click'),
('user3', 'purchase'),
('user2', 'click'),
('user1', 'view'),
('user2', 'purchase'),
]
print("Input:")
for k, v in records:
print(f" {k}: {v}")
# Build stream
stream = KStream(records.copy())
# Filter clicks
clicks = stream.filter(lambda k, v: v == 'click')
print("\nFiltered (clicks):")
for k, v in clicks.records:
print(f" {k}: {v}")
# Count per user
counts = KStream(records.copy()).groupByKey().count()
print("\nCounts per user:")
counts.show()
# Word count example
print("\n=== Word Count Example ===")
# Input: sentences
sentences = [
('doc1', 'hello world'),
('doc2', 'hello kafka'),
('doc3', 'world of streams'),
]
# FlatMap to words
words = []
for doc_id, sentence in sentences:
for word in sentence.split():
words.append((word, 1))
print("Words:")
for k, v in words:
print(f" {k}: {v}")
# Count
word_counts = KStream(words).groupByKey().count()
print("\nWord counts:")
word_counts.show()
# STREAMS BEST PRACTICES
print("\n=== Kafka Streams Best Practices ===")
print("""
1. USE EXACTLY-ONCE
- processing.guarantee=exactly_once_v2
- No duplicates
2. HANDLE STATE
- Local state stores
- Changelog topics
- Restore on failure
3. WINDOWING
- Tumbling, hopping, sliding
- Session windows
- Grace period
4. TOPOLOGY DESIGN
- Minimize repartitioning
- Use KTable for lookups
- Avoid unnecessary shuffles
5. MONITORING
- Lag
- State store size
- Processing rate
6. TESTING
- TopologyTestDriver
- Unit tests
- Integration tests
7. SCALING
- Add instances
- Partitions = max parallelism
- Rebalance
8. ERROR HANDLING
- Deserialization errors
- Production exceptions
- Dead letter queue
""")
# STREAMS SUMMARY
print("\n=== Kafka Streams Summary ===")
print(f"{'Concept':<25} {'Description':<40}")
print("-" * 65)
print(f"{'KStream':<25} {'Stream of events':<40}")
print(f"{'KTable':<25} {'Table of latest values':<40}")
print(f"{'Topology':<25} {'Processing graph':<40}")
print(f"{'State Store':<25} {'Local storage':<40}")
print(f"{'Changelog':<25} {'Backed by Kafka topic':<40}")
print(f"{'Exactly-once':<25} {'Transactions':<40}") ✅ Exactly-Once Semantics
🎯 No Duplicates, No Loss
Exactly-once semantics (EOS) ensures each message is processed exactly once, even with failures. Kafka supports EOS through idempotent producers and transactions.
Exactly-Once
Idempotent: No duplicate writes. Transactions: Atomic operations. Producer: enable.idempotence. Consumer: read_committed. Streams: exactly_once_v2. Guarantee: No loss, no duplicates.
✅ Idempotent✅ Transactions✅ No Duplicates✅ No Loss
# Exactly-Once Semantics
print("=== Delivery Guarantees ===")
print("""
1. AT-MOST-ONCE
- Messages may be lost
- No duplicates
- Fastest
- acks=0
2. AT-LEAST-ONCE
- Messages not lost
- May have duplicates
- Default
- acks=all + retries
3. EXACTLY-ONCE
- Messages not lost
- No duplicates
- Slowest
- Idempotent + transactions
""")
print("\n=== Idempotent Producer ===")
print("""
Idempotent Producer:
- No duplicate writes
- Per partition
- Sequence numbers
- Broker deduplication
Enable:
enable.idempotence=true
acks=all
retries=Integer.MAX_VALUE
max.in.flight.requests.per.connection=5
How it works:
- Producer assigns sequence number
- Broker tracks last sequence
- Duplicate → discarded
- Guarantees exactly-once per partition
""")
print("\n=== Transactions ===")
print("""
Transactions:
- Atomic writes across partitions
- All or nothing
- Read committed isolation
Producer:
producer.initTransactions();
producer.beginTransaction();
producer.send(record1);
producer.send(record2);
producer.commitTransaction();
Consumer:
isolation.level=read_committed
(Only read committed messages)
Use cases:
- Stream processing
- Exactly-once pipelines
- Atomic multi-partition writes
""")
print("\n=== Exactly-Once in Kafka Streams ===")
print("""
Kafka Streams:
processing.guarantee=exactly_once_v2
How it works:
- Transactions
- State store updates
- Offset commits
- All atomic
Benefits:
- No duplicates
- No loss
- End-to-end exactly-once
Configuration:
processing.guarantee=exactly_once_v2
(or exactly_once for older versions)
""")
# IMPLEMENTATION
print("\n=== Simulating Exactly-Once ===")
class IdempotentProducer:
"""Simulate idempotent producer."""
def __init__(self, producer_id):
self.producer_id = producer_id
self.sequence = 0
self.sent = []
self.duplicates = 0
def send(self, topic, partition, message):
"""Send with sequence number."""
record = {
'producer_id': self.producer_id,
'sequence': self.sequence,
'topic': topic,
'partition': partition,
'message': message,
}
self.sent.append(record)
self.sequence += 1
return record
class IdempotentBroker:
"""Simulate idempotent broker."""
def __init__(self):
# producer_id -> partition -> last_sequence
self.sequences = {}
self.messages = []
self.duplicates = 0
def append(self, record):
"""Append record with deduplication."""
pid = record['producer_id']
partition = record['partition']
seq = record['sequence']
key = (pid, partition)
if key in self.sequences:
last_seq = self.sequences[key]
if seq <= last_seq:
# Duplicate
self.duplicates += 1
print(f" Duplicate detected: producer={pid}, seq={seq}")
return False
self.sequences[key] = seq
self.messages.append(record)
return True
class TransactionalProducer:
"""Simulate transactional producer."""
def __init__(self, transactional_id):
self.transactional_id = transactional_id
self.in_transaction = False
self.pending = []
def init_transactions(self):
print(f" Producer {self.transactional_id}: Initialized")
def begin_transaction(self):
self.in_transaction = True
self.pending = []
print(f" Producer {self.transactional_id}: BEGIN")
def send(self, topic, message):
if not self.in_transaction:
raise Exception("No transaction")
self.pending.append({'topic': topic, 'message': message})
print(f" Producer {self.transactional_id}: Pending {message}")
def commit_transaction(self, broker):
if not self.in_transaction:
raise Exception("No transaction")
# Commit all
for record in self.pending:
broker.append(record)
print(f" Producer {self.transactional_id}: COMMIT ({len(self.pending)} messages)")
self.in_transaction = False
self.pending = []
def abort_transaction(self):
print(f" Producer {self.transactional_id}: ABORT")
self.in_transaction = False
self.pending = []
# Demo
print("=== Exactly-Once Demo ===")
# Idempotent producer
print("\n1. Idempotent Producer:")
producer = IdempotentProducer('producer-1')
broker = IdempotentBroker()
# Send messages
for i in range(5):
record = producer.send('orders', 0, f'message-{i}')
broker.append(record)
print(f" Sent: {len(producer.sent)}")
print(f" Appended: {len(broker.messages)}")
print(f" Duplicates: {broker.duplicates}")
# Simulate duplicate
print("\n2. Simulating Duplicate:")
record = producer.send('orders', 0, 'duplicate')
# Simulate retry (same sequence)
record['sequence'] = 4 # Old sequence
broker.append(record)
print(f" Total appended: {len(broker.messages)}")
print(f" Duplicates detected: {broker.duplicates}")
# Transactional producer
print("\n3. Transactional Producer:")
tx_producer = TransactionalProducer('tx-producer-1')
tx_producer.init_transactions()
tx_producer.begin_transaction()
tx_producer.send('orders', 'order-1')
tx_producer.send('orders', 'order-2')
tx_producer.send('payments', 'payment-1')
tx_producer.commit_transaction(broker)
print(f" Total messages: {len(broker.messages)}")
# Abort transaction
print("\n4. Aborted Transaction:")
tx_producer.begin_transaction()
tx_producer.send('orders', 'order-3')
tx_producer.send('orders', 'order-4')
tx_producer.abort_transaction()
print(f" Total messages (unchanged): {len(broker.messages)}")
# EOS SUMMARY
print("\n=== Exactly-Once Summary ===")
print(f"{'Guarantee':<25} {'Description':<40}")
print("-" * 65)
print(f"{'At-most-once':<25} {'May lose messages':<40}")
print(f"{'At-least-once':<25} {'May duplicate':<40}")
print(f"{'Exactly-once':<25} {'No loss, no duplicates':<40}")
print("\n=== Configuration ===")
print(f"{'Setting':<35} {'Value':<25}")
print("-" * 60)
print(f"{'enable.idempotence':<35} {'true':<25}")
print(f"{'acks':<35} {'all':<25}")
print(f"{'isolation.level':<35} {'read_committed':<25}")
print(f"{'processing.guarantee':<35} {'exactly_once_v2':<25}")⚡ Performance & Tuning
🎯 Making Kafka Fast
Kafka performance depends on proper tuning of producers, consumers, brokers, and topics. Key factors: batching, compression, partition count, and replication.
Performance Tuning
Throughput: Messages/sec. Latency: End-to-end time. Batching: Group messages. Compression: Reduce size. Partitions: Parallelism. Replication: Durability vs speed.
⚡ Throughput⚡ Latency⚡ Batching⚡ Compression
# Kafka Performance Tuning
print("=== Performance Factors ===")
print("""
1. THROUGHPUT
- Messages per second
- Bytes per second
- Affected by:
- Batching
- Compression
- Partition count
- Replication
2. LATENCY
- End-to-end time
- Affected by:
- Batching (linger.ms)
- Acks
- Network
- Disk
3. TRADE-OFFS
- Throughput vs latency
- Durability vs speed
- Cost vs performance
""")
print("\n=== Producer Tuning ===")
print("""
For throughput:
batch.size=1048576 (1 MB)
linger.ms=100
compression.type=lz4
acks=1
max.in.flight.requests.per.connection=5
For latency:
batch.size=16384
linger.ms=0
compression.type=none
acks=1
For durability:
acks=all
enable.idempotence=true
max.in.flight.requests.per.connection=5
""")
print("\n=== Consumer Tuning ===")
print("""
For throughput:
fetch.min.bytes=1048576 (1 MB)
fetch.max.wait.ms=500
max.partition.fetch.bytes=1048576
max.poll.records=1000
For latency:
fetch.min.bytes=1
fetch.max.wait.ms=10
max.poll.records=100
For reliability:
enable.auto.commit=false
isolation.level=read_committed
""")
print("\n=== Broker Tuning ===")
print("""
Key configurations:
num.network.threads=8
num.io.threads=16
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576
socket.request.max.bytes=104857600
log.segment.bytes=1073741824
log.retention.ms=604800000
log.cleanup.policy=delete
num.partitions=6
default.replication.factor=3
min.insync.replicas=2
num.replica.fetchers=4
replica.fetch.max.bytes=1048576
""")
print("\n=== Topic Tuning ===")
print("""
Partition count:
- Based on throughput
- Target: 10 MB/s per partition
- Rule: 1-10 per broker core
- Max: 200,000 per cluster
Replication factor:
- 3 for production
- 2 for less critical
- 1 for dev
Retention:
- Time-based: 7 days
- Size-based: 1 TB
- Compaction: for KTable
Segment size:
- 1 GB default
- Smaller for faster cleanup
- Larger for fewer files
""")
print("\n=== Compression ===")
print(f"{'Codec':<15} {'Speed':<15} {'Ratio':<15} {'Use Case':<25}")
print("-" * 70)
print(f"{'none':<15} {'Fastest':<15} {'1:1':<15} {'Low latency':<25}")
print(f"{'lz4':<15} {'Fast':<15} {'2:1':<15} {'Balanced':<25}")
print(f"{'snappy':<15} {'Fast':<15} {'2:1':<15} {'Balanced':<25}")
print(f"{'gzip':<15} {'Slow':<15} {'3:1':<15} {'High compression':<25}")
print(f"{'zstd':<15} {'Medium':<15} {'4:1':<15} {'Best ratio':<25}")
# IMPLEMENTATION
print("\n=== Simulating Performance ===")
import time
import random
class KafkaPerformanceSimulator:
"""Simulate Kafka performance."""
def __init__(self, num_partitions=3, replication_factor=3):
self.num_partitions = num_partitions
self.replication_factor = replication_factor
self.metrics = {
'messages_sent': 0,
'bytes_sent': 0,
'batches': 0,
}
def simulate_producer(self, num_messages, batch_size, linger_ms,
compression, acks):
"""Simulate producer performance."""
print(f"\n=== Producer Simulation ===")
print(f"Messages: {num_messages}")
print(f"Batch size: {batch_size}")
print(f"Linger: {linger_ms}ms")
print(f"Compression: {compression}")
print(f"Acks: {acks}")
# Calculate batches
messages_per_batch = batch_size // 100 # Assume 100 bytes/message
num_batches = (num_messages + messages_per_batch - 1) // messages_per_batch
# Calculate time
# Base time per batch
batch_time = 0.001 # 1ms
# Compression factor
compression_factor = {
'none': 1.0,
'lz4': 0.6,
'snappy': 0.6,
'gzip': 0.4,
'zstd': 0.3,
}.get(compression, 1.0)
# Acks factor
acks_factor = {
0: 0.5,
1: 0.8,
'all': 1.2,
}.get(acks, 1.0)
# Calculate total time
total_time = num_batches * batch_time * compression_factor * acks_factor
total_time += linger_ms / 1000.0
# Throughput
throughput = num_messages / total_time
print(f"Batches: {num_batches}")
print(f"Time: {total_time*1000:.1f}ms")
print(f"Throughput: {throughput:.0f} msg/s")
return {
'time': total_time,
'throughput': throughput,
'batches': num_batches,
}
def simulate_consumer(self, num_messages, fetch_size, max_poll):
"""Simulate consumer performance."""
print(f"\n=== Consumer Simulation ===")
print(f"Messages: {num_messages}")
print(f"Fetch size: {fetch_size} bytes")
print(f"Max poll: {max_poll}")
# Calculate polls
num_polls = (num_messages + max_poll - 1) // max_poll
# Time per poll
poll_time = 0.005 # 5ms
total_time = num_polls * poll_time
throughput = num_messages / total_time
print(f"Polls: {num_polls}")
print(f"Time: {total_time*1000:.1f}ms")
print(f"Throughput: {throughput:.0f} msg/s")
return {
'time': total_time,
'throughput': throughput,
'polls': num_polls,
}
# Demo
print("=== Performance Simulation ===")
sim = KafkaPerformanceSimulator()
# Producer scenarios
print("\n### Producer Scenarios ###")
# Low latency
sim.simulate_producer(100000, batch_size=16384, linger_ms=0,
compression='none', acks=1)
# Balanced
sim.simulate_producer(100000, batch_size=1048576, linger_ms=100,
compression='lz4', acks=1)
# High throughput
sim.simulate_producer(100000, batch_size=1048576, linger_ms=100,
compression='lz4', acks=0)
# Durable
sim.simulate_producer(100000, batch_size=1048576, linger_ms=100,
compression='lz4', acks='all')
# Consumer scenarios
print("\n### Consumer Scenarios ###")
sim.simulate_consumer(100000, fetch_size=1048576, max_poll=1000)
sim.simulate_consumer(100000, fetch_size=1048576, max_poll=100)
# PERFORMANCE SUMMARY
print("\n=== Performance Summary ===")
print(f"{'Scenario':<25} {'Config':<40} {'Throughput':<15}")
print("-" * 80)
print(f"{'Low latency':<25} {'small batch, no compression':<40} {'~1M msg/s':<15}")
print(f"{'Balanced':<25} {'1MB batch, lz4':<40} {'~5M msg/s':<15}")
print(f"{'High throughput':<25} {'1MB batch, lz4, acks=0':<40} {'~10M msg/s':<15}")
print(f"{'Durable':<25} {'1MB batch, lz4, acks=all':<40} {'~3M msg/s':<15}")💼 Applications
🎯 Kafka in the Real World
Kafka powers real-time data pipelines, event-driven architectures, and streaming analytics at companies like LinkedIn, Netflix, and Uber. It's the backbone of modern data infrastructure.
Applications
Messaging: Pub/sub. Logs: Aggregation. Streaming: Real-time. Microservices: Events. Analytics: Pipelines. Integration: Connect.
💼 Messaging💼 Logs💼 Streaming💼 Microservices
# Kafka Applications
print("=== 1. Messaging ===")
print("""
Traditional messaging:
- Decouple producers/consumers
- Async communication
- Buffering
- Load leveling
Example:
Order service → Kafka → Email service
Order service → Kafka → Inventory service
Order service → Kafka → Shipping service
Benefits:
- Decoupling
- Scalability
- Fault tolerance
""")
print("\n=== 2. Log Aggregation ===")
print("""
Centralized logging:
- Collect logs from services
- Store in Kafka
- Process with Streams
- Send to Elasticsearch
Pipeline:
Services → Kafka → Logstash → Elasticsearch → Kibana
Benefits:
- Centralized
- Real-time
- Scalable
- Durable
""")
print("\n=== 3. Streaming Analytics ===")
print("""
Real-time analytics:
- Clickstream analysis
- Fraud detection
- Monitoring
- Alerting
Pipeline:
Events → Kafka → Kafka Streams → Dashboard
→ Alerting
→ Database
Example:
-- Count clicks per second
stream
.windowedBy(TimeWindows.of(Duration.ofSeconds(1)))
.count()
.toStream()
.to("click-counts");
""")
print("\n=== 4. Event-Driven Microservices ===")
print("""
Event-driven architecture:
- Services communicate via events
- Loose coupling
- Async
- Scalable
Example:
Order Service:
- Emits: OrderCreated, OrderShipped
- Consumes: PaymentProcessed
Payment Service:
- Consumes: OrderCreated
- Emits: PaymentProcessed
Shipping Service:
- Consumes: PaymentProcessed
- Emits: OrderShipped
Benefits:
- Decoupling
- Resilience
- Scalability
""")
print("\n=== 5. Data Integration ===")
print("""
Kafka Connect:
- Database → Kafka (CDC)
- Kafka → Database
- Kafka → S3/HDFS
- Kafka → Elasticsearch
Pipeline:
PostgreSQL → Debezium → Kafka → HDFS
MySQL → JDBC Source → Kafka → S3
Kafka → HDFS Sink → Hive
Benefits:
- No code
- Scalable
- Fault-tolerant
""")
print("\n=== 6. Metrics and Monitoring ===")
print("""
Metrics pipeline:
- Collect metrics
- Aggregate
- Alert
- Visualize
Pipeline:
Services → Kafka → Kafka Streams → Prometheus
→ Grafana
→ PagerDuty
Example:
-- Alert on high error rate
stream
.filter((k, v) -> v.status >= 500)
.windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
.count()
.filter((k, v) -> v > 100)
.to("alerts");
""")
# IMPLEMENTATION
print("\n=== Practical Example: Order Pipeline ===")
class OrderEvent:
"""Order event."""
def __init__(self, order_id, user_id, amount, status):
self.order_id = order_id
self.user_id = user_id
self.amount = amount
self.status = status
self.timestamp = time.time()
def to_dict(self):
return {
'order_id': self.order_id,
'user_id': self.user_id,
'amount': self.amount,
'status': self.status,
'timestamp': self.timestamp,
}
class OrderPipeline:
"""Simulate order processing pipeline."""
def __init__(self):
self.events = []
self.processed = []
def produce(self, event):
"""Produce order event."""
self.events.append(event)
print(f" Produced: Order {event.order_id} ({event.status})")
def process(self):
"""Process events."""
for event in self.events:
# Validate
if event.amount <= 0:
print(f" Invalid order: {event.order_id}")
continue
# Route based on status
if event.status == 'created':
# Send to payment
self.processed.append({
'event': event.to_dict(),
'action': 'process_payment',
})
print(f" → Payment for order {event.order_id}")
elif event.status == 'paid':
# Send to shipping
self.processed.append({
'event': event.to_dict(),
'action': 'ship_order',
})
print(f" → Shipping for order {event.order_id}")
elif event.status == 'shipped':
# Send to notification
self.processed.append({
'event': event.to_dict(),
'action': 'notify_customer',
})
print(f" → Notification for order {event.order_id}")
# Demo
print("=== Order Pipeline Demo ===")
pipeline = OrderPipeline()
# Create orders
orders = [
OrderEvent(1, 'user1', 100, 'created'),
OrderEvent(2, 'user2', 200, 'created'),
OrderEvent(3, 'user1', 150, 'paid'),
OrderEvent(4, 'user3', 300, 'paid'),
OrderEvent(5, 'user2', 250, 'shipped'),
OrderEvent(6, 'user4', -50, 'created'), # Invalid
]
print("Producing orders:")
for order in orders:
pipeline.produce(order)
print("\nProcessing orders:")
pipeline.process()
print(f"\nTotal processed: {len(pipeline.processed)}")
# APPLICATION SUMMARY
print("\n=== Application Summary ===")
print(f"{'Domain':<25} {'Use Case':<30} {'Scale':<20}")
print("-" * 75)
print(f"{'Messaging':<25} {'Pub/sub':<30} {'Millions/sec':<20}")
print(f"{'Logs':<25} {'Aggregation':<30} {'TBs/day':<20}")
print(f"{'Streaming':<25} {'Analytics':<30} {'Millions/sec':<20}")
print(f"{'Microservices':<25} {'Events':<30} {'Millions/sec':<20}")
print(f"{'Integration':<25} {'Connect':<30} {'Petabytes':<20}")
print(f"{'Monitoring':<25} {'Metrics':<30} {'Millions/sec':<20}")🚀 Future: Kafka in Modern Computing
🎯 Where Kafka Is Going
Kafka is evolving: KRaft, tiered storage, serverless, and integration with AI/ML. It's becoming the central nervous system of modern data infrastructure.
Future Directions
KRaft: No ZooKeeper. Tiered: Cloud storage. Serverless: No ops. AI: ML pipelines. Streams: Better processing. Edge: IoT.
🚀 KRaft🚀 Tiered Storage🚀 Serverless🚀 AI
# Future of Apache Kafka
print("=== 1. KRaft (Kafka Raft) ===")
print("""
KRaft:
- Kafka without ZooKeeper
- Built-in metadata management
- Simpler deployment
- Better scalability
- Kafka 3.x+ (production ready)
Benefits:
- Fewer components
- Easier operations
- Better performance
- Faster failover
Status:
- ZooKeeper: Deprecated
- KRaft: Recommended
- Migration path available
Future:
- KRaft default
- ZooKeeper removed
- Simpler clusters
""")
print("\n=== 2. Tiered Storage ===")
print("""
Tiered Storage:
- Separate compute and storage
- Local + cloud storage
- Infinite retention
- Lower cost
How it works:
- Recent data: Local disk
- Older data: Cloud (S3, GCS)
- Transparent to clients
Benefits:
- Lower cost
- Infinite retention
- Better scalability
- Cloud-native
Configuration:
remote.log.storage.system.enable=true
remote.log.storage.manager.class=...
Future:
- Standard feature
- Multi-cloud
- Cost optimization
""")
print("\n=== 3. Serverless Kafka ===")
print("""
Serverless Kafka:
- No infrastructure
- Pay per use
- Auto-scaling
- Managed
Options:
- Confluent Cloud
- AWS MSK Serverless
- Azure Event Hubs
- Google Cloud Pub/Sub
Benefits:
- No ops
- Elastic
- Cost-effective
- Focus on apps
Future:
- Serverless default
- Event-driven
- Integration with FaaS
""")
print("\n=== 4. AI and ML ===")
print("""
Kafka for AI/ML:
Use cases:
- Real-time feature store
- Model serving
- Online learning
- Data pipelines
Integration:
- TensorFlow
- PyTorch
- MLflow
- Feature stores
Example:
Events → Kafka → Feature engineering → Model
→ Predictions → Kafka → Actions
Future:
- Real-time ML
- Feature stores
- MLOps
""")
print("\n=== 5. Edge and IoT ===")
print("""
Kafka at the edge:
Use cases:
- IoT data collection
- Edge processing
- Real-time decisions
- Offline capability
Options:
- Kafka on edge devices
- Lightweight clients
- Store and forward
Example:
Sensors → Edge Kafka → Cloud Kafka → Analytics
Future:
- Edge-native
- 5G integration
- Real-time everything
""")
print("\n=== 6. Stream Processing Evolution ===")
print("""
Kafka Streams evolution:
Current:
- Java/Scala
- DSL
- State stores
Future:
- More languages
- Better performance
- SQL (ksqlDB)
- Integration with Flink
ksqlDB:
- SQL on streams
- No code
- Real-time
Future:
- Unified processing
- SQL + code
- Multi-language
""")
# TIMELINE
print("\n=== Timeline ===")
print("""
Past:
2011: Kafka created (LinkedIn)
2012: Apache incubator
2014: Kafka 0.8
2016: Kafka Streams
2017: Kafka 0.11 (EOS)
2019: Kafka 2.3 (KRaft)
2021: Kafka 3.0
Present:
2024: KRaft default
2025: Tiered storage
Future:
2030: Serverless default?
2035: AI-native Kafka?
2040: New paradigms?
""")
print("\n=== Final Thoughts ===")
print("""
Apache Kafka is:
- The central nervous system
- Real-time data platform
- Event streaming
- Still evolving
The future is:
- KRaft (no ZooKeeper)
- Tiered storage
- Serverless
- AI/ML
- Edge
- Unified
"Kafka is the backbone
of modern data infrastructure."
Keep exploring. Keep building.
""")