1# docker-compose.yml
2version: '3'
3services:
4  zookeeper:
5    image: confluentinc/cp-zookeeper:7.4.0
6    environment:
7      ZOOKEEPER_CLIENT_PORT: 2181
8    ports:
9      - "2181:2181"
10 
11  kafka:
12    image: confluentinc/cp-kafka:7.4.0
13    depends_on:
14      - zookeeper
15    ports:
16      - "9092:9092"
17    environment:
18      KAFKA_BROKER_ID: 1
19      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
20      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
21      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

1# Start Kafka
2docker-compose up -d
3 
4# Verify
5docker-compose ps

1pip install kafka-python
2# or
3pip install confluent-kafka  # More performant

1from kafka import KafkaProducer
2import json
3
4# Create producer
5producer = KafkaProducer(
6    bootstrap_servers=['localhost:9092'],
7    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
8    key_serializer=lambda k: k.encode('utf-8') if k else None
9)
10
11# Send message (async)
12future = producer.send(
13    topic='my-topic',
14    value={'user_id': 123, 'action': 'click'},
15    key='user_123'
16)
17
18# Wait for confirmation (optional)
19result = future.get(timeout=10)
20print(f"Sent to partition {result.partition} at offset {result.offset}")
21
22# Flush and close
23producer.flush()
24producer.close()

1from kafka import KafkaProducer
2import json
3
4def on_send_success(record_metadata):
5    print(f"Success: topic={record_metadata.topic}, "
6          f"partition={record_metadata.partition}, "
7          f"offset={record_metadata.offset}")
8
9def on_send_error(ex):
10    print(f"Error: {ex}")
11
12producer = KafkaProducer(
13    bootstrap_servers=['localhost:9092'],
14    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
15    acks='all',  # Wait for all replicas
16    retries=3,   # Retry on failure
17    linger_ms=10,  # Batch messages for 10ms
18    batch_size=16384  # Batch size in bytes
19)
20
21# Send with callbacks
22for i in range(100):
23    message = {'id': i, 'data': f'message_{i}'}
24    producer.send('my-topic', value=message) \
25        .add_callback(on_send_success) \
26        .add_errback(on_send_error)
27
28producer.flush()

1producer = KafkaProducer(
2    bootstrap_servers=['localhost:9092'],
3    
4    # Serialization
5    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
6    key_serializer=lambda k: k.encode('utf-8') if k else None,
7    
8    # Reliability
9    acks='all',           # 0, 1, 'all'
10    retries=5,
11    retry_backoff_ms=100,
12    
13    # Performance
14    batch_size=16384,     # Bytes
15    linger_ms=5,          # Wait time for batching
16    buffer_memory=33554432,  # Total buffer memory
17    
18    # Compression
19    compression_type='gzip',  # none, gzip, snappy, lz4, zstd
20    
21    # Idempotence (exactly-once)
22    enable_idempotence=True
23)

1from kafka import KafkaConsumer
2import json
3
4# Create consumer
5consumer = KafkaConsumer(
6    'my-topic',
7    bootstrap_servers=['localhost:9092'],
8    auto_offset_reset='earliest',  # Start from beginning
9    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
10    group_id='my-consumer-group'
11)
12
13# Poll messages
14for message in consumer:
15    print(f"Topic: {message.topic}")
16    print(f"Partition: {message.partition}")
17    print(f"Offset: {message.offset}")
18    print(f"Key: {message.key}")
19    print(f"Value: {message.value}")
20    print("---")
21
22consumer.close()

1from kafka import KafkaConsumer
2
3consumer = KafkaConsumer(
4    'my-topic',
5    bootstrap_servers=['localhost:9092'],
6    auto_offset_reset='earliest',
7    enable_auto_commit=False,  # Manual commit
8    group_id='my-consumer-group'
9)
10
11try:
12    for message in consumer:
13        # Process message
14        process_message(message)
15        
16        # Commit offset after successful processing
17        consumer.commit()
18        
19except Exception as e:
20    print(f"Error: {e}")
21finally:
22    consumer.close()

1# Consumer Group: Multiple consumers sharing work
2# Each partition assigned to exactly one consumer in group
3
4"""
5Topic with 4 partitions:
6┌────────────────────────────────────────┐
7│  Partition 0  │  Partition 1  │        │
8│     ↓         │     ↓         │        │
9│  Consumer 1   │  Consumer 1   │        │
10├───────────────┼───────────────┤        │
11│  Partition 2  │  Partition 3  │        │
12│     ↓         │     ↓         │        │
13│  Consumer 2   │  Consumer 2   │        │
14└────────────────────────────────────────┘
15
162 consumers, 4 partitions = 2 partitions each
17
18If Consumer 2 fails:
19Consumer 1 gets all 4 partitions (rebalancing)
20"""
21
22# Multiple consumers with same group_id
23consumer1 = KafkaConsumer('my-topic', group_id='my-group', ...)
24consumer2 = KafkaConsumer('my-topic', group_id='my-group', ...)
25# Partitions distributed between consumer1 and consumer2

1consumer = KafkaConsumer(
2    'my-topic',
3    bootstrap_servers=['localhost:9092'],
4    
5    # Consumer group
6    group_id='my-consumer-group',
7    
8    # Offset management
9    auto_offset_reset='earliest',  # earliest, latest, none
10    enable_auto_commit=True,
11    auto_commit_interval_ms=5000,
12    
13    # Polling
14    max_poll_records=500,
15    max_poll_interval_ms=300000,
16    
17    # Session
18    session_timeout_ms=10000,
19    heartbeat_interval_ms=3000,
20    
21    # Deserialization
22    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
23    key_deserializer=lambda m: m.decode('utf-8') if m else None
24)

1from kafka.admin import KafkaAdminClient, NewTopic
2
3admin = KafkaAdminClient(bootstrap_servers=['localhost:9092'])
4
5# Create topic
6topic = NewTopic(
7    name='new-topic',
8    num_partitions=3,
9    replication_factor=1,
10    topic_configs={
11        'retention.ms': '86400000',  # 1 day
12        'cleanup.policy': 'delete'
13    }
14)
15
16admin.create_topics([topic])
17
18# List topics
19topics = admin.list_topics()
20print(topics)
21
22# Delete topic
23admin.delete_topics(['old-topic'])
24
25admin.close()

1# Create topic
2kafka-topics.sh --create --topic my-topic \
3  --bootstrap-server localhost:9092 \
4  --partitions 3 --replication-factor 1
5 
6# List topics
7kafka-topics.sh --list --bootstrap-server localhost:9092
8 
9# Describe topic
10kafka-topics.sh --describe --topic my-topic \
11  --bootstrap-server localhost:9092
12 
13# Produce messages (CLI)
14kafka-console-producer.sh --topic my-topic \
15  --bootstrap-server localhost:9092
16 
17# Consume messages (CLI)
18kafka-console-consumer.sh --topic my-topic \
19  --bootstrap-server localhost:9092 --from-beginning
20 
21# Consumer groups
22kafka-consumer-groups.sh --list --bootstrap-server localhost:9092
23kafka-consumer-groups.sh --describe --group my-group \
24  --bootstrap-server localhost:9092

1from pyspark.sql import SparkSession
2from pyspark.sql.functions import *
3from pyspark.sql.types import *
4
5spark = SparkSession.builder \
6    .appName("Kafka-Spark") \
7    .config("spark.jars.packages", 
8            "org.apache.spark:spark-sql-kafka-0-10_2.12:3.4.0") \
9    .getOrCreate()
10
11# Read from Kafka
12df = spark.readStream \
13    .format("kafka") \
14    .option("kafka.bootstrap.servers", "localhost:9092") \
15    .option("subscribe", "my-topic") \
16    .option("startingOffsets", "earliest") \
17    .load()
18
19# Parse JSON
20schema = StructType([
21    StructField("user_id", StringType()),
22    StructField("action", StringType()),
23    StructField("timestamp", TimestampType())
24])
25
26parsed = df \
27    .selectExpr("CAST(value AS STRING)") \
28    .select(from_json(col("value"), schema).alias("data")) \
29    .select("data.*")
30
31# Process and write back to Kafka
32output = parsed \
33    .groupBy(window(col("timestamp"), "5 minutes"), col("action")) \
34    .count()
35
36query = output \
37    .selectExpr("to_json(struct(*)) AS value") \
38    .writeStream \
39    .format("kafka") \
40    .option("kafka.bootstrap.servers", "localhost:9092") \
41    .option("topic", "processed-topic") \
42    .option("checkpointLocation", "/checkpoints/kafka") \
43    .start()
44
45query.awaitTermination()

1# Producer: Order Service
2def publish_order_event(order):
3    event = {
4        'event_type': 'ORDER_CREATED',
5        'order_id': order['id'],
6        'customer_id': order['customer_id'],
7        'items': order['items'],
8        'total': order['total'],
9        'timestamp': datetime.now().isoformat()
10    }
11    producer.send('orders', key=str(order['id']), value=event)
12
13# Consumer: Inventory Service
14def process_order_for_inventory(message):
15    event = message.value
16    if event['event_type'] == 'ORDER_CREATED':
17        for item in event['items']:
18            reserve_inventory(item['product_id'], item['quantity'])
19
20# Consumer: Notification Service
21def process_order_for_notification(message):
22    event = message.value
23    if event['event_type'] == 'ORDER_CREATED':
24        send_confirmation_email(event['customer_id'], event['order_id'])
25
26# Each service has its own consumer group
27inventory_consumer = KafkaConsumer('orders', group_id='inventory-service')
28notification_consumer = KafkaConsumer('orders', group_id='notification-service')
29# Both receive all messages (different groups)

1from kafka import KafkaProducer, KafkaConsumer
2import json
3
4producer = KafkaProducer(
5    bootstrap_servers=['localhost:9092'],
6    value_serializer=lambda v: json.dumps(v).encode('utf-8')
7)
8
9consumer = KafkaConsumer(
10    'main-topic',
11    bootstrap_servers=['localhost:9092'],
12    group_id='my-group',
13    enable_auto_commit=False
14)
15
16MAX_RETRIES = 3
17
18for message in consumer:
19    retries = 0
20    while retries < MAX_RETRIES:
21        try:
22            process_message(message.value)
23            consumer.commit()
24            break
25        except Exception as e:
26            retries += 1
27            if retries == MAX_RETRIES:
28                # Send to Dead Letter Queue
29                producer.send(
30                    'main-topic-dlq',
31                    value={
32                        'original_message': message.value,
33                        'error': str(e),
34                        'original_topic': message.topic,
35                        'original_partition': message.partition,
36                        'original_offset': message.offset
37                    }
38                )
39                consumer.commit()

1# Key-based partitioning (default)
2# Same key always goes to same partition
3producer.send('orders', key='customer_123', value=order)
4# All orders for customer_123 in same partition = ordered
5
6# Custom partitioner
7from kafka.partitioner import Partitioner
8
9class CustomPartitioner(Partitioner):
10    def partition(self, topic, key, all_partitions, available_partitions):
11        # Custom logic
12        if key:
13            return hash(key) % len(all_partitions)
14        return 0
15
16producer = KafkaProducer(partitioner=CustomPartitioner())

1# Producer metrics
2metrics = producer.metrics()
3print(metrics)
4
5# Consumer lag
6from kafka import KafkaAdminClient
7from kafka.admin import ConsumerGroupCommand
8
9# Check consumer lag via CLI
10# kafka-consumer-groups.sh --describe --group my-group \
11#   --bootstrap-server localhost:9092

1from kafka.errors import KafkaError, NoBrokersAvailable
2
3def safe_produce(producer, topic, value, retries=3):
4    for attempt in range(retries):
5        try:
6            future = producer.send(topic, value=value)
7            result = future.get(timeout=10)
8            return result
9        except NoBrokersAvailable:
10            print(f"No brokers available, retry {attempt + 1}")
11            time.sleep(1)
12        except KafkaError as e:
13            print(f"Kafka error: {e}")
14            if attempt == retries - 1:
15                raise
16    raise Exception("Failed to produce after retries")

1# Scenario: Real-time clickstream processing
2# 
3# 1. Producer: Generate click events
4#    - user_id, page_url, timestamp, session_id
5#
6# 2. Consumer 1: Count clicks per page (last 5 min)
7#
8# 3. Consumer 2: Detect suspicious activity
9#    - Alert if user has >100 clicks in 1 minute
10#
11# Requirements:
12# - Use consumer groups
13# - Handle errors with DLQ
14# - Include proper serialization
15
16# YOUR CODE HERE