protocols-kafka-streams

Apache Kafka stream processing and event streaming platform

Kafka Streams

Scope: Kafka architecture, producers, consumers, Kafka Streams API, exactly-once semantics, schema registry, ksqlDB Lines: ~300 Last Updated: 2025-10-27

When to Use This Skill

Activate this skill when:

Core Concepts

What is Apache Kafka?

Apache Kafka: Distributed event streaming platform

Use cases:

Architecture Components

Topic: Logical stream of records (like a table)

Partition: Ordered sequence of messages within topic

Broker: Kafka server instance

Producer: Publishes messages to topics Consumer: Reads messages from topics Consumer Group: Set of consumers sharing topic partitions

Producer → Topic (Partitions) → Consumer Group → Consumers
                   ↓
            Replication (ISR)

Basic Producer/Consumer

Producer (Python)

from kafka import KafkaProducer
import json

# Create producer
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',  # Wait for all in-sync replicas
    retries=3,
    compression_type='lz4'
)

# Send message
producer.send(
    'orders',
    key='order-123',
    value={'order_id': 'order-123', 'amount': 99.99}
)

# Flush and close
producer.flush()
producer.close()

Consumer (Python)

from kafka import KafkaConsumer
import json

# Create consumer
consumer = KafkaConsumer(
    'orders',
    bootstrap_servers=['localhost:9092'],
    group_id='order-processor',
    value_deserializer=lambda m: json.loads(m.decode('utf-8')),
    auto_offset_reset='earliest',
    enable_auto_commit=False  # Manual commit for reliability
)

# Consume messages
for message in consumer:
    print(f"Partition: {message.partition}, Offset: {message.offset}")
    print(f"Value: {message.value}")

    # Process message
    process_order(message.value)

    # Commit after processing (at-least-once)
    consumer.commit()

consumer.close()

Kafka Streams API

Stream Processing (Java)

import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.*;

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "word-count-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

StreamsBuilder builder = new StreamsBuilder();

// Input topic
KStream<String, String> textLines = builder.stream("text-input");

// Process: split words → count
KTable<String, Long> wordCounts = textLines
    .flatMapValues(line -> Arrays.asList(line.split("\\s+")))
    .groupBy((key, word) -> word)
    .count();

// Output topic
wordCounts.toStream().to("word-count-output");

// Start streams app
KafkaStreams streams = new KafkaStreams(builder.build(), props);
streams.start();

Stateless Operations

// Filter
KStream<String, Integer> evens = stream.filter((k, v) -> v % 2 == 0);

// Map
KStream<String, String> upper = stream.mapValues(v -> v.toUpperCase());

// FlatMap
KStream<String, String> words = stream.flatMapValues(
    line -> Arrays.asList(line.split("\\s+"))
);

// Branch (split stream)
KStream<String, Integer>[] branches = stream.branch(
    (k, v) -> v > 0,   // Positive
    (k, v) -> v < 0,   // Negative
    (k, v) -> true     // Zero
);

Stateful Operations

// Group and count
KTable<String, Long> counts = stream
    .groupByKey()
    .count();

// Aggregate
KTable<String, Integer> sums = stream
    .groupByKey()
    .reduce((aggValue, newValue) -> aggValue + newValue);

// Custom aggregation
KTable<String, Stats> stats = stream
    .groupByKey()
    .aggregate(
        Stats::new,  // Initializer
        (key, value, aggregate) -> {
            aggregate.sum += value;
            aggregate.count++;
            return aggregate;
        }
    );

Windowing

// Tumbling window (5 minutes)
KTable<Windowed<String>, Long> counts = stream
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .count();

// Hopping window
KTable<Windowed<String>, Long> counts = stream
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(10))
                           .advanceBy(Duration.ofMinutes(5)))
    .count();

// Session window (30-minute inactivity gap)
KTable<Windowed<String>, Long> sessions = stream
    .groupByKey()
    .windowedBy(SessionWindows.with(Duration.ofMinutes(30)))
    .count();

Exactly-Once Semantics

Transactional Producer (Python)

from kafka import KafkaProducer

# Create transactional producer
producer = KafkaProducer(
    transactional_id='order-service-1',  # Unique ID
    enable_idempotence=True,
    acks='all'
)

# Initialize transactions
producer.init_transactions()

try:
    # Begin transaction
    producer.begin_transaction()

    # Send messages (atomic across topics)
    producer.send('orders', order_data)
    producer.send('payments', payment_data)

    # Commit transaction
    producer.commit_transaction()

except Exception as e:
    # Rollback on error
    producer.abort_transaction()

Exactly-Once Streams

// Enable exactly-once in Kafka Streams
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
          StreamsConfig.EXACTLY_ONCE_V2);

// Streams app now has exactly-once guarantees:
// - Output writes are transactional
// - State stores are consistent
// - Offset commits are atomic

Schema Registry (Avro)

Producer with Avro

from confluent_kafka import avro
from confluent_kafka.avro import AvroProducer

# Define schema
user_schema = avro.loads('''
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "id", "type": "string"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": "string"},
    {"name": "age", "type": ["null", "int"], "default": null}
  ]
}
''')

# Create producer
producer = AvroProducer({
    'bootstrap.servers': 'localhost:9092',
    'schema.registry.url': 'http://localhost:8081'
}, default_value_schema=user_schema)

# Send message (Avro serialization automatic)
user = {
    'id': 'user-1',
    'name': 'Alice',
    'email': 'alice@example.com',
    'age': 30
}
producer.produce(topic='users', value=user)
producer.flush()

Consumer with Avro

from confluent_kafka.avro import AvroConsumer

consumer = AvroConsumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'user-consumer',
    'schema.registry.url': 'http://localhost:8081'
})

consumer.subscribe(['users'])

while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue

    # Value automatically deserialized
    user = msg.value()
    print(f"User: {user['name']} ({user['email']})")

ksqlDB

Basic Queries

-- Create stream from topic
CREATE STREAM user_events (
  user_id VARCHAR,
  action VARCHAR,
  timestamp BIGINT
) WITH (
  KAFKA_TOPIC='user-events',
  VALUE_FORMAT='JSON'
);

-- Query stream (continuous)
SELECT * FROM user_events EMIT CHANGES;

-- Create table (aggregation)
CREATE TABLE event_counts AS
SELECT user_id, COUNT(*) AS count
FROM user_events
GROUP BY user_id
EMIT CHANGES;

-- Query table (point-in-time)
SELECT * FROM event_counts WHERE user_id='user123';

Windowing

-- Tumbling window (5 minutes)
CREATE TABLE event_counts_5min AS
SELECT
  user_id,
  COUNT(*) AS count,
  WINDOWSTART AS window_start
FROM user_events
WINDOW TUMBLING (SIZE 5 MINUTES)
GROUP BY user_id
EMIT CHANGES;

-- Session window (30-minute gap)
CREATE TABLE user_sessions AS
SELECT user_id, COUNT(*) AS event_count
FROM user_events
WINDOW SESSION (30 MINUTES)
GROUP BY user_id
EMIT CHANGES;

Joins

-- Stream-stream join
CREATE STREAM order_payments AS
SELECT
  o.order_id,
  o.amount,
  p.payment_id,
  p.status
FROM orders o
INNER JOIN payments p WITHIN 1 HOUR
ON o.order_id = p.order_id
EMIT CHANGES;

-- Stream-table join (enrichment)
CREATE STREAM enriched_orders AS
SELECT
  o.order_id,
  o.amount,
  u.name AS user_name,
  u.email AS user_email
FROM orders o
LEFT JOIN users u ON o.user_id = u.user_id
EMIT CHANGES;

Performance Tuning

Producer Tuning

High throughput:

producer = KafkaProducer(
    acks=1,  # Leader only
    compression_type='lz4',
    batch_size=32768,  # 32 KB
    linger_ms=20,  # Wait to batch
    buffer_memory=67108864  # 64 MB
)

High reliability:

producer = KafkaProducer(
    acks='all',  # All ISR
    enable_idempotence=True,
    retries=3,
    max_in_flight_requests_per_connection=5
)

Consumer Tuning

High throughput:

consumer = KafkaConsumer(
    fetch_min_bytes=1048576,  # 1 MB
    max_poll_records=1000
)

Low latency:

consumer = KafkaConsumer(
    fetch_min_bytes=1,
    fetch_max_wait_ms=0
)

Monitoring

Key Metrics

Broker health:

Producer:

Consumer:

Consumer Lag

# Check consumer lag
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-group --describe

# Output:
# TOPIC    PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG
# orders   0          1000            1500            500
# orders   1          2000            2000            0

Thresholds:

Common Patterns

Partitioning Strategies

Key-based (ordering guarantee):

producer.send('orders', key='user123', value=data)
# Same key → same partition → ordering preserved

Custom partitioner:

from kafka.partitioner import Partitioner

class VIPPartitioner(Partitioner):
    def partition(self, key, all_partitions, available):
        if key.startswith('vip_'):
            return 0  # VIP users to partition 0
        return hash(key) % len(all_partitions)

Offset Management

Auto-commit (at-most-once):

consumer = KafkaConsumer(enable_auto_commit=True)
# Simple, but can lose data on crash

Manual commit (at-least-once):

consumer = KafkaConsumer(enable_auto_commit=False)
for msg in consumer:
    process(msg)
    consumer.commit()  # Commit after processing

Batch commit (throughput):

for i, msg in enumerate(consumer):
    process(msg)
    if i % 100 == 0:
        consumer.commit()

Anti-Patterns

❌ Creating producer per message: Expensive connection overhead ✅ Create once, reuse for all messages

❌ Not setting timeouts: Can hang forever ✅ Always set request_timeout_ms, timeout parameters

❌ Large messages (> 1 MB): Performance issues ✅ Split into smaller messages or use compression

❌ No replication: Data loss on broker failure ✅ Use replication.factor=3, min.insync.replicas=2

❌ Too many partitions: Overhead, slow rebalancing ✅ Start with 10-100 partitions per topic

❌ Ignoring consumer lag: Consumers falling behind ✅ Monitor lag, alert on thresholds, scale consumers

Level 3: Resources

Overview

This skill includes comprehensive Level 3 resources for deep Kafka knowledge and practical automation tools.

Resources include:

REFERENCE.md

Location: skills/protocols/kafka-streams/resources/REFERENCE.md

Comprehensive technical reference (2,438 lines) covering:

Core Topics:

Key Sections:

  1. Fundamentals: Topics, partitions, brokers, replication, ISR
  2. Producer API: Configuration, partitioning, batching, compression
  3. Consumer API: Groups, offset management, rebalancing
  4. Kafka Streams: Topology, stateless/stateful ops, state stores
  5. Windowing: Tumbling, hopping, sliding, session windows
  6. Joins: Stream-stream, stream-table, table-table joins
  7. Exactly-Once: Transactional producer, idempotent writes, EOS v2
  8. Schema Registry: Avro schemas, backward/forward compatibility
  9. ksqlDB: SQL queries, push/pull queries, materialized views
  10. Performance: Throughput vs latency tuning, partition sizing
  11. Monitoring: Key metrics, alerting thresholds, lag monitoring
  12. Security: Authentication (SASL), encryption (TLS), authorization (ACLs)

Format: Markdown with extensive code examples in Python, Java

Scripts

Three production-ready executable scripts in resources/scripts/:

1. validate_kafka_config.py (588 lines)

Purpose: Validate Kafka broker and topic configurations against best practices

Features:

Usage:

# Validate entire cluster
./validate_kafka_config.py --bootstrap-servers localhost:9092

# Validate specific topics
./validate_kafka_config.py --bootstrap-servers localhost:9092 --topics orders,payments

# JSON output for CI/CD
./validate_kafka_config.py --bootstrap-servers localhost:9092 --json > report.json

Categories checked:

2. analyze_consumer_lag.py (537 lines)

Purpose: Monitor and analyze consumer lag across consumer groups

Features:

Usage:

# Analyze all consumer groups
./analyze_consumer_lag.py --bootstrap-servers localhost:9092

# Analyze specific group
./analyze_consumer_lag.py --bootstrap-servers localhost:9092 --group order-processor

# Custom lag threshold
./analyze_consumer_lag.py --bootstrap-servers localhost:9092 --threshold 5000

# JSON output for monitoring systems
./analyze_consumer_lag.py --bootstrap-servers localhost:9092 --json

Metrics:

Thresholds:

3. benchmark_throughput.py (620 lines)

Purpose: Benchmark Kafka producer/consumer throughput and latency

Features:

Usage:

# Producer benchmark (100k messages)
./benchmark_throughput.py --bootstrap-servers localhost:9092 --mode producer --messages 100000

# Consumer benchmark
./benchmark_throughput.py --bootstrap-servers localhost:9092 --mode consumer --topic orders

# End-to-end benchmark (producer + consumer)
./benchmark_throughput.py --bootstrap-servers localhost:9092 --mode both --messages 50000

# Custom message size (10 KB)
./benchmark_throughput.py --bootstrap-servers localhost:9092 --mode producer --messages 10000 --size 10240

# JSON output
./benchmark_throughput.py --bootstrap-servers localhost:9092 --mode producer --messages 10000 --json

Metrics:

Examples

Nine production-ready examples in resources/examples/:

1. python/basic_producer.py (200 lines)

Complete Python producer demonstrating:

2. python/basic_consumer.py (280 lines)

Complete Python consumer demonstrating:

3. python/exactly_once_producer.py (180 lines)

Transactional producer with exactly-once semantics:

4. python/avro_schema_registry.py (220 lines)

Avro serialization with Schema Registry:

5. java/WordCountStreamsApp.java (200 lines)

Kafka Streams application (Java):

6. ksqldb/queries.sql (500 lines)

Comprehensive ksqlDB query examples:

7. docker/docker-compose.yml (280 lines)

Complete Kafka stack with Docker Compose:

Services:

8. monitoring/prometheus_config.yml (250 lines)

Prometheus configuration for Kafka monitoring:

Key alerts:

9. examples/README.md (420 lines)

Comprehensive guide with:

Quick Start

1. Validate Kafka cluster:

cd skills/protocols/kafka-streams/resources/scripts
./validate_kafka_config.py --bootstrap-servers localhost:9092 --json

2. Monitor consumer lag:

./analyze_consumer_lag.py --bootstrap-servers localhost:9092 --group my-group

3. Benchmark throughput:

./benchmark_throughput.py --bootstrap-servers localhost:9092 --mode both --messages 10000

4. Run examples with Docker:

cd ../examples/docker
docker-compose up -d

# Wait for services
docker-compose logs -f kafka

# Access Control Center: http://localhost:9021

5. Test producer/consumer:

cd ../examples/python
python basic_producer.py
python basic_consumer.py  # In another terminal

File Structure

skills/protocols/kafka-streams/
├── kafka-streams.md (this file)
└── resources/
    ├── REFERENCE.md (2,438 lines)
    ├── scripts/
    │   ├── validate_kafka_config.py (588 lines)
    │   ├── analyze_consumer_lag.py (537 lines)
    │   └── benchmark_throughput.py (620 lines)
    └── examples/
        ├── README.md (420 lines)
        ├── python/
        │   ├── basic_producer.py (200 lines)
        │   ├── basic_consumer.py (280 lines)
        │   ├── exactly_once_producer.py (180 lines)
        │   └── avro_schema_registry.py (220 lines)
        ├── java/
        │   └── WordCountStreamsApp.java (200 lines)
        ├── ksqldb/
        │   └── queries.sql (500 lines)
        ├── docker/
        │   └── docker-compose.yml (280 lines)
        └── monitoring/
            └── prometheus_config.yml (250 lines)

Resources Summary

| Category | Item | Lines | Description | |----------|------|-------|-------------| | Reference | REFERENCE.md | 2,438 | Complete technical reference | | Scripts | validate_kafka_config.py | 588 | Config validator | | | analyze_consumer_lag.py | 537 | Lag analyzer | | | benchmark_throughput.py | 620 | Throughput benchmark | | Examples | basic_producer.py | 200 | Producer patterns | | | basic_consumer.py | 280 | Consumer patterns | | | exactly_once_producer.py | 180 | Transactions | | | avro_schema_registry.py | 220 | Schema Registry | | | WordCountStreamsApp.java | 200 | Kafka Streams | | | queries.sql | 500 | ksqlDB queries | | | docker-compose.yml | 280 | Docker stack | | | prometheus_config.yml | 250 | Monitoring | | | README.md | 420 | Examples guide |

Total: 6,713 lines of production-ready resources

Last Updated: 2025-10-27 Format Version: 1.0 (Atomic)