Streaming Data

Name: Streaming Data
Author: ancoleman

ancoleman/ai-design-components

60 installs
426 repo stars
Updated December 11, 2025
ancoleman/ai-design-components

Streaming-data is a Claude skill that builds event streaming and real-time data pipelines with Kafka, Pulsar, Redpanda, Flink and Spark.

About

This skill builds event streaming systems and real-time data pipelines with brokers (Kafka, Pulsar, Redpanda) and stream processors (Flink, Spark, Kafka Streams). Developers use it for event-driven architectures, real-time analytics, and CDC/ETL integration. It covers broker and processor selection, delivery guarantees, and producer/consumer patterns across multiple languages.

Message brokers vs stream processors and when to use each
Delivery guarantees: at-most-once, at-least-once, exactly-once
Producer/consumer patterns with DLQ and offset management

Streaming Data by the numbers

60 all-time installs (skills.sh)
Ranked #389 of 911 Databases skills by installs in the Skillselion catalog
Data as of Aug 5, 2026 (Skillselion catalog sync)

At a glance

streaming-data capabilities & compatibility

Capabilities: event streaming · stream processing · producer consumer · cdc · data pipelines
Works with: kafka
Use cases: data analysis · api development
Pricing: Free

From the docs

What streaming-data says it does

Build event streaming and real-time data pipelines with Kafka, Pulsar, Redpanda, Flink, and Spark.

SKILL.md

npx skills add https://github.com/ancoleman/ai-design-components --skill streaming-data

Add your badge

Show developers this skill is listed on Skillselion. Paste this into your README.

[![Listed on Skillselion](https://skillselion.com/badge/skills/ancoleman/ai-design-components/streaming-data.svg)](https://skillselion.com/skills/ancoleman/ai-design-components/streaming-data)

Installs	60
repo stars	★ 426
Last updated	December 11, 2025
Repository	ancoleman/ai-design-components ↗

What it does

Build event streaming and real-time data pipelines with Kafka, Pulsar, Redpanda, Flink and Spark.

Who is it for?

Building event-driven architectures and real-time data pipelines.

Skip if: Simple CRUD apps with no real-time or event-streaming requirements.

When should I use this skill?

Building real-time systems, microservices communication, or data integration pipelines.

What you get

Production event streaming with the right broker, processor and delivery guarantee.

Broker and stream-processor selection
Producer/consumer implementations
DLQ, retry and offset-management patterns

By the numbers

Three delivery guarantee models (at-most-once, at-least-once, exactly-once)
Four brokers compared (Kafka, Redpanda, Pulsar, RabbitMQ)

Files

SKILL.mdMarkdownGitHub ↗

Streaming Data Processing

Build production-ready event streaming systems and real-time data pipelines using modern message brokers and stream processors.

When to Use This Skill

Use this skill when:

Building event-driven architectures and microservices communication
Processing real-time analytics, monitoring, or alerting systems
Implementing data integration pipelines (CDC, ETL/ELT)
Creating log or metrics aggregation systems
Developing IoT platforms or high-frequency trading systems

Core Concepts

Message Brokers vs Stream Processors

Message Brokers (Kafka, Pulsar, Redpanda):

Store and distribute event streams
Provide durability, replay capability, partitioning
Handle producer/consumer coordination

Stream Processors (Flink, Spark, Kafka Streams):

Transform and aggregate streaming data
Provide windowing, joins, stateful operations
Execute complex event processing (CEP)

Delivery Guarantees

At-Most-Once:

Messages may be lost, no duplicates
Lowest overhead
Use for: Metrics, logs where loss is acceptable

At-Least-Once:

Messages never lost, may have duplicates
Moderate overhead, requires idempotent consumers
Use for: Most applications (default choice)

Exactly-Once:

Messages never lost or duplicated
Highest overhead, requires transactional processing
Use for: Financial transactions, critical state updates

Quick Start Guide

Step 1: Choose a Message Broker

See references/broker-selection.md for detailed comparison.

Quick decision:

Apache Kafka: Mature ecosystem, enterprise features, event sourcing
Redpanda: Low latency, Kafka-compatible, simpler operations (no ZooKeeper)
Apache Pulsar: Multi-tenancy, geo-replication, tiered storage
RabbitMQ: Traditional message queues, RPC patterns

Step 2: Choose a Stream Processor (if needed)

See references/processor-selection.md for detailed comparison.

Quick decision:

Apache Flink: Millisecond latency, real-time analytics, CEP
Apache Spark: Batch + stream hybrid, ML integration, analytics
Kafka Streams: Embedded in microservices, no separate cluster
ksqlDB: SQL interface for stream processing

Step 3: Implement Producer/Consumer Patterns

Choose language-specific guide:

TypeScript/Node.js: references/typescript-patterns.md (KafkaJS)
Python: references/python-patterns.md (confluent-kafka-python)
Go: references/go-patterns.md (kafka-go)
Java/Scala: references/java-patterns.md (Apache Kafka Java Client)

Common Patterns

Basic Producer Pattern

Send events to a topic with error handling:

1. Create producer with broker addresses
2. Configure delivery guarantees (acks, retries, idempotence)
3. Send messages with key (for partitioning) and value
4. Handle delivery callbacks or errors
5. Flush and close producer on shutdown

Basic Consumer Pattern

Process events from topics with offset management:

1. Create consumer with broker addresses and group ID
2. Subscribe to topics
3. Poll for messages
4. Process each message
5. Commit offsets (auto or manual)
6. Handle errors (retry, DLQ, skip)
7. Close consumer gracefully

Error Handling Strategy

For production systems, implement:

Dead Letter Queue (DLQ): Send failed messages to separate topic
Retry Logic: Configurable retry attempts with backoff
Graceful Shutdown: Finish processing, commit offsets, close connections
Monitoring: Track consumer lag, error rates, throughput

Decision Frameworks

Framework: Message Broker Selection

START: What are requirements?

1. Need Kafka API compatibility?
   YES → Kafka or Redpanda
   NO → Continue

2. Is multi-tenancy critical?
   YES → Apache Pulsar
   NO → Continue

3. Operational simplicity priority?
   YES → Redpanda (single binary, no ZooKeeper)
   NO → Continue

4. Mature ecosystem needed?
   YES → Apache Kafka
   NO → Redpanda (better performance)

5. Task queues (not event streams)?
   YES → RabbitMQ or message-queues skill
   NO → Kafka/Redpanda/Pulsar

Framework: Stream Processor Selection

START: What is latency requirement?

1. Millisecond-level latency needed?
   YES → Apache Flink
   NO → Continue

2. Batch + stream in same pipeline?
   YES → Apache Spark Streaming
   NO → Continue

3. Embedded in microservice?
   YES → Kafka Streams
   NO → Continue

4. SQL interface for analysts?
   YES → ksqlDB
   NO → Flink or Spark

5. Python primary language?
   YES → Spark (PySpark) or Faust
   NO → Flink (Java/Scala)

Framework: Language Selection

TypeScript/Node.js:

API gateways, web services, real-time dashboards
KafkaJS library (827 code snippets, high reputation)

Python:

Data science, ML pipelines, analytics
confluent-kafka-python (192 snippets, score 68.8)

Go:

High-performance microservices, infrastructure tools
kafka-go (42 snippets, idiomatic Go)

Java/Scala:

Enterprise applications, Kafka Streams, Flink, Spark
Apache Kafka Java Client (683 snippets, score 76.9)

Advanced Patterns

Event Sourcing

Store state changes as immutable events. See references/event-sourcing.md for:

Event store design patterns
Event schema evolution
Snapshot strategies
Temporal queries and audit trails

Change Data Capture (CDC)

Capture database changes as events. See references/cdc-patterns.md for:

Debezium integration (MySQL, PostgreSQL, MongoDB)
Real-time data synchronization
Microservices data integration patterns

Exactly-Once Processing

Implement transactional guarantees. See references/exactly-once.md for:

Idempotent producers
Transactional consumers
End-to-end exactly-once pipelines

Error Handling

Production-grade error management. See references/error-handling.md for:

Dead letter queue patterns
Retry strategies with exponential backoff
Backpressure handling
Circuit breakers for downstream failures

Reference Files

Decision Guides

references/broker-selection.md - Kafka vs Pulsar vs Redpanda comparison
references/processor-selection.md - Flink vs Spark vs Kafka Streams
references/delivery-guarantees.md - At-least-once, exactly-once patterns

Language-Specific Implementation

references/typescript-patterns.md - KafkaJS patterns (producer, consumer, error handling)
references/python-patterns.md - confluent-kafka-python patterns
references/go-patterns.md - kafka-go patterns
references/java-patterns.md - Apache Kafka Java client patterns

Advanced Topics

references/event-sourcing.md - Event sourcing architecture
references/cdc-patterns.md - Change Data Capture with Debezium
references/exactly-once.md - Transactional processing
references/error-handling.md - DLQ, retries, backpressure
references/performance-tuning.md - Throughput optimization, partitioning strategies

Validation Scripts

Run these scripts for token-free validation and generation:

Validate Kafka Configuration

python scripts/validate-kafka-config.py --config producer.yaml
python scripts/validate-kafka-config.py --config consumer.yaml

Checks: broker connectivity, configuration validity, serialization format

Generate Schema Registry Templates

python scripts/generate-schema.py --type avro --entity User
python scripts/generate-schema.py --type protobuf --entity Event

Creates: Avro/Protobuf schema definitions for Schema Registry

Benchmark Throughput

bash scripts/benchmark-throughput.sh --broker localhost:9092 --topic test

Tests: Producer/consumer throughput, latency percentiles

Code Examples

TypeScript Example (KafkaJS)

See examples/typescript/ for:

basic-producer.ts - Simple event producer with error handling
basic-consumer.ts - Consumer with manual offset commits
transactional-producer.ts - Exactly-once producer pattern
consumer-with-dlq.ts - Dead letter queue implementation

Python Example (confluent-kafka-python)

See examples/python/ for:

basic_producer.py - Producer with delivery callbacks
basic_consumer.py - Consumer with error handling
async_producer.py - AsyncIO producer (aiokafka)
schema_registry.py - Avro serialization with Schema Registry

Go Example (kafka-go)

See examples/go/ for:

basic_producer.go - Idiomatic Go producer
basic_consumer.go - Consumer with manual commits
high_perf_consumer.go - Concurrent processing pattern
batch_producer.go - Batch message sending

Java Example (Apache Kafka)

See examples/java/ for:

BasicProducer.java - Producer with idempotence
BasicConsumer.java - Consumer with error recovery
TransactionalProducer.java - Exactly-once transactions
StreamsAggregation.java - Kafka Streams aggregation

Technology Comparison

Message Broker Comparison

Feature	Kafka	Pulsar	Redpanda	RabbitMQ
Throughput	Very High	High	Very High	Medium
Latency	Medium	Medium	Low	Low
Event Replay	Yes	Yes	Yes	No
Multi-Tenancy	Manual	Native	Manual	Manual
Operational Complexity	Medium	High	Low	Low
Best For	Enterprise, big data	SaaS, IoT	Performance-critical	Task queues

Stream Processor Comparison

Feature	Flink	Spark	Kafka Streams	ksqlDB
Processing Model	True streaming	Micro-batch	Library	SQL engine
Latency	Millisecond	Second	Millisecond	Second
Deployment	Cluster	Cluster	Embedded	Server
Best For	Real-time analytics	Batch + stream	Microservices	Analysts

Client Library Recommendations

Language	Library	Trust Score	Snippets	Use Case
TypeScript	KafkaJS	High	827	Web services, APIs
Python	confluent-kafka-python	High (68.8)	192	Data pipelines, ML
Go	kafka-go	High	42	High-perf services
Java	Kafka Java Client	High (76.9)	683	Enterprise, Flink/Spark

Related Skills

For authentication and security patterns, see the auth-security skill. For infrastructure deployment (Kubernetes operators, Terraform), see the infrastructure-as-code skill. For monitoring metrics and tracing, see the observability skill. For API design patterns, see the api-design-principles skill. For data architecture and warehousing, see the data-architecture skill.

Troubleshooting

Consumer Lag Issues

Check partition count vs consumer count (match for parallelism)
Increase consumer instances or reduce processing time
Monitor with Kafka consumer lag metrics

Message Loss

Verify producer acks=all configuration
Check broker replication factor (>1)
Ensure consumers commit offsets after processing

Duplicate Messages

Implement idempotent consumers (track message IDs)
Use exactly-once semantics (transactions)
Design for at-least-once delivery

Performance Bottlenecks

Increase partition count for parallelism
Tune batch size and linger time
Enable compression (GZIP, LZ4, Snappy)
See references/performance-tuning.md for details

"""
Basic Kafka Consumer Example (Python/confluent-kafka-python)

Demonstrates:
- Consumer configuration with manual offset commits
- Processing messages with error handling
- Dead-letter queue pattern
- Graceful shutdown

Dependencies:
    pip install confluent-kafka

Usage:
    python basic_consumer.py
"""

from confluent_kafka import Consumer, Producer, KafkaException
import json
import signal
import sys

class BasicConsumer:
    def __init__(self, bootstrap_servers: str, group_id: str):
        """Initialize Kafka consumer with manual offset management."""
        self.config = {
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id,
            'auto.offset.reset': 'earliest',
            # Manual commit for error handling
            'enable.auto.commit': False,
        }
        self.consumer = Consumer(self.config)
        self.running = True

        # DLQ producer
        self.dlq_producer = Producer({
            'bootstrap.servers': bootstrap_servers,
        })

    def subscribe(self, topics: list):
        """Subscribe to topics."""
        self.consumer.subscribe(topics)
        print(f'✓ Subscribed to topics: {topics}')

    def consume(self, handler):
        """Start consuming messages."""
        print('✓ Consumer started, waiting for messages...')

        try:
            while self.running:
                msg = self.consumer.poll(timeout=1.0)

                if msg is None:
                    continue

                if msg.error():
                    raise KafkaException(msg.error())

                try:
                    # Decode message
                    value = json.loads(msg.value().decode('utf-8'))

                    # Process message
                    handler(value)

                    # Commit offset after successful processing
                    self.consumer.commit(message=msg)
                    print(f'✓ Processed and committed offset {msg.offset()}')

                except json.JSONDecodeError as e:
                    print(f'✗ Failed to decode message: {e}')
                    self._send_to_dlq(msg, str(e))
                    self.consumer.commit(message=msg)

                except Exception as e:
                    print(f'✗ Error processing message: {e}')
                    # Don't commit - message will be reprocessed

        except KeyboardInterrupt:
            print('\\n✓ Shutdown signal received')
        finally:
            self.close()

    def _send_to_dlq(self, msg, error: str):
        """Send failed message to dead-letter queue."""
        dlq_topic = f'{msg.topic()}.dlq'

        self.dlq_producer.produce(
            topic=dlq_topic,
            key=msg.key(),
            value=msg.value(),
            headers={
                'original-topic': msg.topic(),
                'error-message': error,
            }
        )
        self.dlq_producer.flush()
        print(f'✓ Sent message to DLQ: {dlq_topic}')

    def close(self):
        """Close the consumer."""
        self.consumer.close()
        self.dlq_producer.flush()
        print('✓ Consumer closed')

    def shutdown(self, signum, frame):
        """Graceful shutdown handler."""
        print('\\n✓ Shutting down gracefully...')
        self.running = False

def handle_event(event: dict):
    """Example event handler."""
    print(f'Processing event: {event}')
    # Your business logic here

if __name__ == '__main__':
    consumer = BasicConsumer('localhost:9092', 'basic-consumer-group')

    # Set up signal handlers
    signal.signal(signal.SIGINT, consumer.shutdown)
    signal.signal(signal.SIGTERM, consumer.shutdown)

    consumer.subscribe(['user-actions'])
    consumer.consume(handle_event)

/**
 * Basic Kafka Producer Example (TypeScript/KafkaJS)
 *
 * Demonstrates:
 * - Producer configuration with at-least-once delivery
 * - Sending messages with keys and headers
 * - Error handling with delivery callbacks
 * - Graceful shutdown
 *
 * Dependencies:
 *   npm install kafkajs
 *
 * Usage:
 *   npx ts-node basic-producer.ts
 */

import { Kafka, CompressionTypes, Partitioners, RecordMetadata } from 'kafkajs';

interface UserEvent {
  userId: string;
  action: string;
  timestamp: number;
}

class BasicProducer {
  private kafka: Kafka;
  private producer: any;

  constructor(brokers: string[]) {
    this.kafka = new Kafka({
      clientId: 'basic-producer-example',
      brokers: brokers,
    });

    this.producer = this.kafka.producer({
      createPartitioner: Partitioners.LegacyPartitioner,
      // At-least-once delivery guarantees
      idempotent: true,
      maxInFlightRequests: 5,
    });
  }

  async connect(): Promise<void> {
    await this.producer.connect();
    console.log('✓ Producer connected');
  }

  async sendEvent(topic: string, event: UserEvent): Promise<void> {
    try {
      const metadata: RecordMetadata[] = await this.producer.send({
        topic,
        compression: CompressionTypes.GZIP,
        messages: [
          {
            key: event.userId,
            value: JSON.stringify(event),
            headers: {
              'event-type': event.action,
              'timestamp': event.timestamp.toString(),
            },
          },
        ],
      });

      console.log(`✓ Event sent to partition ${metadata[0].partition}, offset ${metadata[0].offset}`);
    } catch (error) {
      console.error('✗ Failed to send event:', error);
      throw error;
    }
  }

  async disconnect(): Promise<void> {
    await this.producer.disconnect();
    console.log('✓ Producer disconnected');
  }
}

// Main execution
async function main() {
  const producer = new BasicProducer(['localhost:9092']);

  try {
    await producer.connect();

    // Send some example events
    for (let i = 0; i < 10; i++) {
      await producer.sendEvent('user-actions', {
        userId: `user-${i}`,
        action: 'login',
        timestamp: Date.now(),
      });
    }

    console.log('✓ All events sent successfully');
  } catch (error) {
    console.error('✗ Error:', error);
    process.exit(1);
  } finally {
    await producer.disconnect();
  }
}

// Run if executed directly
if (require.main === module) {
  main().catch(console.error);
}

export { BasicProducer, UserEvent };

skill: "streaming-data"
version: "1.0"
domain: "data"

base_outputs:
  # Core streaming infrastructure configurations
  - path: "streaming/kafka/broker-config.yaml"
    must_contain:
      - "broker.id"
      - "log.dirs"
      - "num.partitions"
      - "replication.factor"

  - path: "streaming/kafka/producer-config.yaml"
    must_contain:
      - "bootstrap.servers"
      - "acks"
      - "retries"
      - "idempotence"

  - path: "streaming/kafka/consumer-config.yaml"
    must_contain:
      - "bootstrap.servers"
      - "group.id"
      - "auto.offset.reset"
      - "enable.auto.commit"

  # Topic management and schemas
  - path: "streaming/topics/topic-definitions.yaml"
    must_contain:
      - "topic_name"
      - "partitions"
      - "replication_factor"
      - "retention_ms"

  - path: "schemas/README.md"
    must_contain:
      - "Schema Registry"
      - "Avro"
      - "schema evolution"

  # Error handling and observability
  - path: "streaming/error-handling/dlq-config.yaml"
    must_contain:
      - "dead_letter_topic"
      - "retry_policy"
      - "max_retries"

  - path: "monitoring/metrics-config.yaml"
    must_contain:
      - "consumer.lag"
      - "producer.record-send-rate"
      - "broker.under-replicated-partitions"

conditional_outputs:
  maturity:
    starter:
      # Basic producer/consumer setup
      - path: "streaming/producers/basic-producer.ts"
        must_contain:
          - "Kafka("
          - "producer.connect()"
          - "producer.send("
          - "idempotent: true"

      - path: "streaming/consumers/basic-consumer.ts"
        must_contain:
          - "consumer.connect()"
          - "consumer.subscribe("
          - "consumer.run("
          - "commitOffsets"

      - path: "streaming/README.md"
        must_contain:
          - "Getting Started"
          - "Producer Pattern"
          - "Consumer Pattern"
          - "At-Least-Once Delivery"

    intermediate:
      # Advanced patterns and stream processing
      - path: "streaming/producers/transactional-producer.ts"
        must_contain:
          - "transaction()"
          - "exactly-once"
          - "transactional.id"

      - path: "streaming/consumers/consumer-with-dlq.ts"
        must_contain:
          - "dead letter queue"
          - "retry logic"
          - "error handling"

      - path: "streaming/processors/kafka-streams-app.java"
        must_contain:
          - "StreamsBuilder"
          - "KStream"
          - "aggregate"
          - "windowing"

      - path: "schemas/avro/user-event.avsc"
        must_contain:
          - "type"
          - "namespace"
          - "fields"

      - path: "streaming/cdc/debezium-connector.json"
        must_contain:
          - "connector.class"
          - "database.hostname"
          - "table.include.list"

    advanced:
      # Production-grade patterns and multi-language support
      - path: "streaming/processors/flink-job.java"
        must_contain:
          - "StreamExecutionEnvironment"
          - "DataStream"
          - "window("
          - "checkpoint"

      - path: "streaming/event-sourcing/event-store.ts"
        must_contain:
          - "event store"
          - "append events"
          - "event versioning"
          - "snapshot"

      - path: "streaming/exactly-once/transactional-pipeline.ts"
        must_contain:
          - "exactly-once semantics"
          - "transaction coordinator"
          - "commit marker"

      - path: "streaming/performance/partitioning-strategy.ts"
        must_contain:
          - "custom partitioner"
          - "partition key"
          - "load balancing"

      - path: "monitoring/grafana-dashboard.json"
        must_contain:
          - "consumer lag"
          - "throughput"
          - "latency percentiles"

  queue:
    kafka:
      - path: "streaming/kafka/docker-compose.yaml"
        must_contain:
          - "image: confluentinc/cp-kafka"
          - "KAFKA_BROKER_ID"
          - "KAFKA_ZOOKEEPER_CONNECT"
          - "KAFKA_ADVERTISED_LISTENERS"

      - path: "streaming/kafka/client-config.properties"
        must_contain:
          - "bootstrap.servers"
          - "security.protocol"
          - "compression.type"

      - path: "streaming/kafka/schema-registry-config.yaml"
        must_contain:
          - "kafkastore.connection.url"
          - "schema.registry.url"

    pulsar:
      - path: "streaming/pulsar/broker-config.conf"
        must_contain:
          - "zookeeperServers"
          - "brokerServicePort"
          - "managedLedgerDefaultEnsembleSize"

      - path: "streaming/pulsar/producer-config.yaml"
        must_contain:
          - "topic"
          - "producerName"
          - "sendTimeout"
          - "batchingEnabled"

      - path: "streaming/pulsar/tenant-namespace.yaml"
        must_contain:
          - "tenant"
          - "namespace"
          - "retention_policies"

    redpanda:
      - path: "streaming/redpanda/redpanda.yaml"
        must_contain:
          - "kafka_api:"
          - "admin:"
          - "pandaproxy:"
          - "data_directory:"

      - path: "streaming/redpanda/docker-compose.yaml"
        must_contain:
          - "image: vectorized/redpanda"
          - "redpanda start"
          - "--kafka-addr"

      - path: "streaming/redpanda/rpk-config.yaml"
        must_contain:
          - "brokers:"
          - "tls:"
          - "sasl:"

    rabbitmq:
      - path: "streaming/rabbitmq/rabbitmq.conf"
        must_contain:
          - "listeners.tcp"
          - "default_user"
          - "default_vhost"

      - path: "streaming/rabbitmq/producer-consumer.ts"
        must_contain:
          - "amqplib"
          - "channel.sendToQueue"
          - "channel.consume"
          - "channel.ack"

  cloud_provider:
    aws:
      - path: "streaming/aws/msk-cluster.tf"
        must_contain:
          - "aws_msk_cluster"
          - "kafka_version"
          - "number_of_broker_nodes"

      - path: "streaming/aws/kinesis-stream.tf"
        must_contain:
          - "aws_kinesis_stream"
          - "shard_count"
          - "retention_period"

      - path: "streaming/aws/lambda-consumer.ts"
        must_contain:
          - "KinesisStreamEvent"
          - "event.Records"
          - "kinesis.data"

    gcp:
      - path: "streaming/gcp/pubsub-topic.tf"
        must_contain:
          - "google_pubsub_topic"
          - "google_pubsub_subscription"
          - "message_retention_duration"

      - path: "streaming/gcp/dataflow-pipeline.py"
        must_contain:
          - "apache_beam"
          - "ReadFromPubSub"
          - "WriteToBigQuery"

    azure:
      - path: "streaming/azure/eventhub-namespace.tf"
        must_contain:
          - "azurerm_eventhub_namespace"
          - "azurerm_eventhub"
          - "partition_count"

      - path: "streaming/azure/stream-analytics-job.json"
        must_contain:
          - "input"
          - "output"
          - "transformation"

  language:
    typescript:
      - path: "streaming/typescript/basic-producer.ts"
        must_contain:
          - "kafkajs"
          - "producer.send("
          - "CompressionTypes"

      - path: "streaming/typescript/basic-consumer.ts"
        must_contain:
          - "consumer.run("
          - "eachMessage"
          - "heartbeat()"

      - path: "streaming/typescript/package.json"
        must_contain:
          - "kafkajs"
          - "@types/node"

    python:
      - path: "streaming/python/basic_producer.py"
        must_contain:
          - "confluent_kafka"
          - "Producer("
          - "produce("
          - "flush()"

      - path: "streaming/python/basic_consumer.py"
        must_contain:
          - "Consumer("
          - "subscribe("
          - "poll("
          - "commit()"

      - path: "streaming/python/requirements.txt"
        must_contain:
          - "confluent-kafka"
          - "avro-python3"

    go:
      - path: "streaming/go/basic_producer.go"
        must_contain:
          - "github.com/segmentio/kafka-go"
          - "kafka.Writer"
          - "WriteMessages"

      - path: "streaming/go/basic_consumer.go"
        must_contain:
          - "kafka.Reader"
          - "ReadMessage"
          - "CommitMessages"

      - path: "streaming/go/go.mod"
        must_contain:
          - "module"
          - "github.com/segmentio/kafka-go"

    java:
      - path: "streaming/java/BasicProducer.java"
        must_contain:
          - "org.apache.kafka.clients.producer"
          - "KafkaProducer"
          - "send("
          - "ProducerRecord"

      - path: "streaming/java/BasicConsumer.java"
        must_contain:
          - "org.apache.kafka.clients.consumer"
          - "KafkaConsumer"
          - "poll("
          - "commitSync()"

      - path: "streaming/java/pom.xml"
        must_contain:
          - "kafka-clients"
          - "org.apache.kafka"

scaffolding:
  # Directory structure for streaming systems
  - path: "streaming/"
    reason: "Root directory for all streaming infrastructure and code"

  - path: "streaming/kafka/"
    reason: "Kafka-specific configurations, docker-compose, and setup files"

  - path: "streaming/producers/"
    reason: "Producer implementations for various patterns (basic, transactional, batch)"

  - path: "streaming/consumers/"
    reason: "Consumer implementations including DLQ, retry logic, and error handling"

  - path: "streaming/processors/"
    reason: "Stream processing applications (Flink, Spark, Kafka Streams, ksqlDB)"

  - path: "streaming/topics/"
    reason: "Topic definitions, partitioning strategies, and retention policies"

  - path: "schemas/"
    reason: "Schema definitions for Avro, Protobuf, and JSON Schema"

  - path: "schemas/avro/"
    reason: "Avro schema definitions for Schema Registry"

  - path: "schemas/protobuf/"
    reason: "Protobuf schema definitions for type-safe serialization"

  - path: "streaming/cdc/"
    reason: "Change Data Capture configurations for Debezium connectors"

  - path: "streaming/event-sourcing/"
    reason: "Event sourcing patterns, event store implementations, and snapshots"

  - path: "streaming/exactly-once/"
    reason: "Transactional processing implementations for exactly-once semantics"

  - path: "streaming/error-handling/"
    reason: "Dead letter queue configs, retry policies, and circuit breakers"

  - path: "streaming/performance/"
    reason: "Performance tuning configs, custom partitioners, and benchmarks"

  - path: "monitoring/"
    reason: "Metrics exporters, Grafana dashboards, and alerting rules"

  - path: "tests/"
    reason: "Integration tests for producers, consumers, and stream processors"

metadata:
  primary_blueprints:
    - "data-pipeline"

  contributes_to:
    - "Stream processing infrastructure"
    - "Event-driven architecture"
    - "Real-time data pipelines"
    - "Microservices communication"
    - "CDC and event sourcing"
    - "IoT data ingestion"

  integrates_with:
    - "observability" # Metrics, tracing, and monitoring
    - "auth-security" # SASL/SSL authentication
    - "infrastructure-as-code" # Terraform/K8s deployment
    - "transforming-data" # Downstream data transformation
    - "data-architecture" # Data lake/warehouse integration

  common_patterns:
    - "Producer/Consumer pattern"
    - "Dead Letter Queue (DLQ)"
    - "At-least-once delivery"
    - "Exactly-once processing"
    - "Event sourcing"
    - "Change Data Capture (CDC)"
    - "Stream joins and windowing"
    - "Backpressure handling"

  technology_stack:
    message_brokers:
      - "Apache Kafka"
      - "Apache Pulsar"
      - "Redpanda"
      - "RabbitMQ"
      - "AWS MSK/Kinesis"
      - "GCP Pub/Sub"
      - "Azure Event Hubs"

    stream_processors:
      - "Apache Flink"
      - "Apache Spark Streaming"
      - "Kafka Streams"
      - "ksqlDB"
      - "Faust (Python)"

    client_libraries:
      - "KafkaJS (TypeScript)"
      - "confluent-kafka-python"
      - "kafka-go"
      - "Apache Kafka Java Client"

    serialization:
      - "Apache Avro"
      - "Protocol Buffers"
      - "JSON Schema"
      - "Schema Registry"

    cdc_tools:
      - "Debezium"
      - "Maxwell"
      - "AWS DMS"

  validation:
    scripts:
      - "validate-kafka-config.py - Validates broker/producer/consumer configs"
      - "generate-schema.py - Generates Avro/Protobuf schemas"
      - "benchmark-throughput.sh - Tests producer/consumer performance"

    checks:
      - "Broker connectivity validation"
      - "Topic existence and partition count"
      - "Consumer group status and lag"
      - "Schema Registry compatibility"
      - "Serialization format validation"

  examples_structure:
    typescript:
      - "basic-producer.ts - Simple event producer with error handling"
      - "basic-consumer.ts - Consumer with manual offset commits"
      - "transactional-producer.ts - Exactly-once producer pattern"
      - "consumer-with-dlq.ts - Dead letter queue implementation"

    python:
      - "basic_producer.py - Producer with delivery callbacks"
      - "basic_consumer.py - Consumer with error handling"
      - "async_producer.py - AsyncIO producer (aiokafka)"
      - "schema_registry.py - Avro serialization with Schema Registry"

    go:
      - "basic_producer.go - Idiomatic Go producer"
      - "basic_consumer.go - Consumer with manual commits"
      - "high_perf_consumer.go - Concurrent processing pattern"
      - "batch_producer.go - Batch message sending"

    java:
      - "BasicProducer.java - Producer with idempotence"
      - "BasicConsumer.java - Consumer with error recovery"
      - "TransactionalProducer.java - Exactly-once transactions"
      - "StreamsAggregation.java - Kafka Streams aggregation"

Overview

Choose a message broker based on throughput requirements, latency constraints, operational complexity, and ecosystem maturity.

Apache Kafka

Architecture

Partitioned log-based storage
Consumer groups for load balancing
ZooKeeper dependency (KRaft mode available)
Distributed, fault-tolerant

Strengths

Very high throughput (millions of messages/sec)
Durability and event replay capability
Massive ecosystem (Kafka Connect, Schema Registry, ksqlDB)
Exactly-once semantics support
Battle-tested at scale

Weaknesses

Operational complexity (JVM tuning, ZooKeeper management)
Higher tail latency under load vs alternatives
Resource-intensive (memory, disk, network)

Best Use Cases

Event sourcing and CQRS architectures
Data pipeline integration (150+ Kafka Connect connectors)
High-throughput batch workloads (fintech, analytics)
Enterprise systems with mature tooling requirements
Log and metrics aggregation

Configuration Recommendations

Broker Settings:

# Replication for fault tolerance
replication.factor=3
min.insync.replicas=2

# Performance tuning
num.network.threads=8
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576

Producer Settings:

# Exactly-once
enable.idempotence=true
acks=all
retries=Integer.MAX_VALUE
max.in.flight.requests.per.connection=5

# Performance
compression.type=lz4
batch.size=32768
linger.ms=10

Apache Pulsar

Architecture

Layered architecture (brokers + BookKeeper storage)
Separate compute and storage layers
Native multi-tenancy support
Tiered storage (hot/cold data separation)

Strengths

Excellent multi-tenancy isolation
Geo-replication and cross-datacenter sync
Independent scaling of compute and storage
Schema evolution built-in
Pulsar Functions (lightweight stream processing)

Weaknesses

Higher operational complexity (ZooKeeper + BookKeeper)
Smaller ecosystem compared to Kafka
More moving parts (brokers, bookies, ZooKeeper)

Best Use Cases

Multi-tenant SaaS platforms
IoT platforms with millions of topics
Cross-region data synchronization
Applications requiring tiered storage
Dynamic scaling requirements

Configuration Recommendations

Broker Settings:

# Multi-tenancy
numTenants=1000
maxTopicsPerNamespace=10000

# Tiered storage
managedLedgerOffloadDriver=aws-s3
s3ManagedLedgerOffloadBucket=pulsar-offload

Producer Settings:

Producer<byte[]> producer = client.newProducer()
    .topic("persistent://tenant/namespace/topic")
    .batchingMaxMessages(1000)
    .compressionType(CompressionType.LZ4)
    .create();

Redpanda

Architecture

Single-binary deployment (C++ implementation)
Raft consensus (no ZooKeeper dependency)
Kafka-compatible API
Thread-per-core design for CPU efficiency

Strengths

Lower latency than Kafka (especially tail latency)
Simpler operations (no JVM, no ZooKeeper)
Better CPU and memory utilization
Drop-in Kafka replacement (API compatible)
Fewer nodes needed (cost savings)

Weaknesses

Smaller ecosystem than Kafka
Less mature tooling
Newer project (less battle-tested)

Best Use Cases

Performance-critical applications (low-latency requirements)
Edge computing and resource-constrained environments
Kafka replacements seeking operational simplicity
Cost optimization (fewer nodes for same throughput)
Greenfield projects with performance focus

Configuration Recommendations

Broker Settings:

# redpanda.yaml
redpanda:
  data_directory: /var/lib/redpanda/data
  node_id: 1
  rpc_server:
    address: 0.0.0.0
    port: 33145
  kafka_api:
    - address: 0.0.0.0
      port: 9092
  admin:
    - address: 0.0.0.0
      port: 9644

# Performance tuning
pandaproxy_client:
  retries: 10
  retry_base_backoff_ms: 100

RabbitMQ

Architecture

Queue-based (not log-based)
AMQP, MQTT, STOMP protocol support
Flexible routing (exchanges, bindings)
Message acknowledgements

Strengths

Flexible message routing patterns
Priority queues and message TTL
Easy to set up and operate
Rich plugin ecosystem
Good for RPC patterns

Weaknesses

No event replay capability
Lower throughput than Kafka/Pulsar/Redpanda
Not designed for event streaming use cases

Best Use Cases

Task queues and job processing
RPC communication patterns
Traditional message queue use cases
Microservices async communication (non-streaming)

Configuration Recommendations

RabbitMQ Config:

# rabbitmq.conf
vm_memory_high_watermark.relative = 0.6
disk_free_limit.absolute = 50GB
consumer_timeout = 3600000

# Clustering
cluster_formation.peer_discovery_backend = rabbit_peer_discovery_k8s

Comparison Matrix

Performance Characteristics

Feature	Kafka	Pulsar	Redpanda	RabbitMQ
Throughput	Very High (100k+ msg/s)	High (50k+ msg/s)	Very High (100k+ msg/s)	Medium (10k-50k msg/s)
Latency (p99)	20-100ms	20-100ms	5-50ms	5-20ms
Event Replay	Yes	Yes	Yes	No
Persistence	Disk (log segments)	BookKeeper	Disk (log segments)	Disk/Memory
Retention	Time/Size-based	Time/Size-based	Time/Size-based	Queue-based

Operational Characteristics

Feature	Kafka	Pulsar	Redpanda	RabbitMQ
Deployment Complexity	Medium	High	Low	Low
Dependencies	ZooKeeper (or KRaft)	ZooKeeper + BookKeeper	None (Raft)	None
Resource Usage	High (JVM)	High	Low (C++)	Medium
Scaling	Add brokers	Independent compute/storage	Add brokers	Add nodes
Monitoring	JMX, Prometheus	Prometheus	Prometheus	Management UI

Ecosystem Maturity

Feature	Kafka	Pulsar	Redpanda	RabbitMQ
Client Libraries	Excellent	Good	Kafka-compatible	Excellent
Connectors	150+ (Kafka Connect)	Good (Pulsar IO)	Kafka-compatible	Plugin-based
Stream Processing	Kafka Streams, ksqlDB	Pulsar Functions	Kafka-compatible	Limited
Schema Registry	Confluent Schema Registry	Built-in	Compatible	N/A
Community Size	Very Large	Medium	Growing	Large

Cost Considerations

Factor	Kafka	Pulsar	Redpanda	RabbitMQ
Hardware Requirements	High	High	Medium	Low-Medium
Node Count	3-5+ brokers	3+ brokers + bookies	3+ brokers	3+ nodes
Operational Overhead	Medium	High	Low	Low
Cloud Pricing	$$$	$$$	$$	$

Selection Flowchart

Primary Decision Path

START: What is primary use case?

├─ Event Streaming & Event Sourcing
│  ├─ Need proven ecosystem? → KAFKA
│  ├─ Need lowest latency? → REDPANDA
│  └─ Need multi-tenancy? → PULSAR
│
├─ Real-Time Analytics
│  ├─ Millisecond latency? → REDPANDA
│  └─ Integration with big data? → KAFKA
│
├─ Data Integration Pipelines
│  ├─ Many source connectors? → KAFKA (Kafka Connect)
│  └─ Cross-region sync? → PULSAR
│
├─ Microservices Communication
│  ├─ Event-driven architecture? → KAFKA or REDPANDA
│  └─ Task queues? → RABBITMQ
│
└─ IoT / Edge Computing
   ├─ Resource-constrained? → REDPANDA
   └─ Millions of topics? → PULSAR

Operational Considerations

START: What are operational constraints?

├─ Team Experience
│  ├─ Strong Kafka expertise? → KAFKA
│  ├─ Need simplicity? → REDPANDA or RABBITMQ
│  └─ Multi-cloud experience? → PULSAR
│
├─ Infrastructure
│  ├─ Kubernetes-native? → REDPANDA or PULSAR
│  ├─ Traditional VMs? → KAFKA or RABBITMQ
│  └─ Edge devices? → REDPANDA
│
└─ Budget
   ├─ Cost-sensitive? → REDPANDA (fewer nodes)
   ├─ Enterprise support needed? → KAFKA (Confluent)
   └─ Open source only? → KAFKA or REDPANDA

Performance Requirements

START: What are performance needs?

├─ Throughput
│  ├─ >100k msg/s per node? → KAFKA or REDPANDA
│  ├─ 50k-100k msg/s? → PULSAR
│  └─ <50k msg/s? → RABBITMQ
│
├─ Latency
│  ├─ <10ms p99? → REDPANDA
│  ├─ <50ms p99? → KAFKA or PULSAR
│  └─ <100ms p99? → RABBITMQ
│
└─ Guarantees
   ├─ Exactly-once critical? → KAFKA
   ├─ At-least-once OK? → ANY
   └─ At-most-once OK? → RABBITMQ

Technology-Specific Guidance

When to Choose Kafka

Strong indicators:

Need for battle-tested, mature ecosystem
Requirement for event replay and time-travel debugging
Large number of data source integrations (Kafka Connect)
Enterprise support requirements (Confluent Platform)
Team already has Kafka expertise

Example scenarios:

Financial transaction processing (exactly-once semantics)
E-commerce event sourcing (order events, inventory changes)
Data lake ingestion (S3, HDFS, data warehouse)
Microservices event-driven architecture

When to Choose Pulsar

Strong indicators:

Multi-tenant SaaS application
Geo-replication across multiple regions
Need to separate compute and storage scaling
Tiered storage for hot/cold data
Millions of topics (IoT scenarios)

Example scenarios:

SaaS platform with tenant isolation
IoT device telemetry (millions of devices)
Cross-region data synchronization
Message routing with complex topic hierarchies

When to Choose Redpanda

Strong indicators:

Low-latency requirements (<10ms p99)
Operational simplicity priority
Cost optimization (fewer nodes)
Kafka compatibility needed (existing clients)
Resource-constrained environments

Example scenarios:

High-frequency trading systems
Real-time fraud detection
Edge computing applications
Kafka replacement for cost/performance
Gaming telemetry (low latency critical)

When to Choose RabbitMQ

Strong indicators:

Task queue processing (not event streaming)
RPC communication patterns
Need for flexible message routing
Priority queues required
Simpler use cases

Example scenarios:

Background job processing
Email sending queues
Request-response patterns
Notification delivery systems

Migration Paths

From RabbitMQ to Kafka/Redpanda

Why migrate:

Need event replay capability
Scaling beyond RabbitMQ throughput limits
Event-driven architecture adoption

Migration strategy: 1. Run both systems in parallel (dual-write) 2. Migrate consumers first (read from Kafka) 3. Migrate producers (write to Kafka) 4. Decommission RabbitMQ

From Kafka to Redpanda

Why migrate:

Reduce operational complexity
Lower latency requirements
Cost optimization

Migration strategy: 1. Redpanda is Kafka API-compatible 2. Point clients to Redpanda brokers 3. Mirror topics using MirrorMaker 2 4. Cutover consumers and producers 5. Decommission Kafka cluster

From Kafka to Pulsar

Why migrate:

Multi-tenancy requirements
Need tiered storage
Geo-replication

Migration strategy: 1. Deploy Pulsar cluster 2. Use Pulsar Kafka-on-Pulsar adapter 3. Mirror topics with Kafka Connect 4. Migrate consumers to Pulsar client 5. Migrate producers 6. Decommission Kafka

Conclusion

Default recommendation: Start with Apache Kafka unless specific requirements dictate otherwise. Kafka offers the best balance of features, maturity, and ecosystem.

Performance-critical: Choose Redpanda for low-latency requirements and operational simplicity.

Multi-tenant SaaS: Choose Pulsar for native multi-tenancy and geo-replication.

Simple queues: Choose RabbitMQ for traditional message queue use cases.

Change Data Capture (CDC) Patterns

Overview
Use Cases
Debezium (Recommended)
MySQL CDC Example
PostgreSQL CDC Example
Consuming CDC Events
Outbox Pattern
Best Practices
Monitoring
Conclusion

Overview

Change Data Capture captures changes from databases and publishes them as events to streaming platforms. Essential for real-time data synchronization and microservices data integration.

Use Cases

Real-time data replication
Microservices data synchronization
Event-driven architectures
Data warehouse ingestion
Cache invalidation

Debezium (Recommended)

Debezium is the industry-standard CDC tool for Kafka. It captures row-level changes from databases and publishes them to Kafka topics.

Supported Databases

MySQL
PostgreSQL
MongoDB
SQL Server
Oracle
Db2
Cassandra

Architecture

Database → Debezium Connector → Kafka → Consumers

MySQL CDC Example

1. Enable Binary Logging

-- MySQL configuration (my.cnf)
server-id = 1
log_bin = mysql-bin
binlog_format = ROW
binlog_row_image = FULL
expire_logs_days = 10

2. Create Debezium User

CREATE USER 'debezium'@'%' IDENTIFIED BY 'password';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT
ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

3. Deploy Debezium Connector

{
  "name": "mysql-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "password",
    "database.server.id": "184054",
    "database.server.name": "mydb",
    "database.include.list": "inventory",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    "database.history.kafka.topic": "schema-changes.inventory"
  }
}

4. Event Format

{
  "before": {
    "id": 1,
    "name": "Old Name",
    "email": "old@example.com"
  },
  "after": {
    "id": 1,
    "name": "New Name",
    "email": "new@example.com"
  },
  "source": {
    "version": "1.9.0.Final",
    "connector": "mysql",
    "name": "mydb",
    "ts_ms": 1234567890,
    "snapshot": "false",
    "db": "inventory",
    "table": "users",
    "server_id": 1,
    "gtid": null,
    "file": "mysql-bin.000003",
    "pos": 154,
    "row": 0
  },
  "op": "u",
  "ts_ms": 1234567890
}

PostgreSQL CDC Example

1. Enable Logical Replication

-- postgresql.conf
wal_level = logical
max_replication_slots = 4
max_wal_senders = 4

2. Create Replication User

CREATE USER debezium WITH REPLICATION PASSWORD 'password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium;

3. Deploy Connector

{
  "name": "postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "password",
    "database.dbname": "inventory",
    "database.server.name": "pgserver",
    "plugin.name": "pgoutput",
    "publication.name": "dbz_publication"
  }
}

Consuming CDC Events

TypeScript Consumer

import { Kafka } from 'kafkajs';

interface CDCEvent {
  before: any;
  after: any;
  source: {
    db: string;
    table: string;
    ts_ms: number;
  };
  op: 'c' | 'u' | 'd' | 'r'; // create, update, delete, read
  ts_ms: number;
}

class CDCConsumer {
  private consumer: Consumer;

  async subscribe(tables: string[]): Promise<void> {
    const topics = tables.map(t => `mydb.inventory.${t}`);
    await this.consumer.subscribe({ topics });
  }

  async consume(): Promise<void> {
    await this.consumer.run({
      eachMessage: async ({ topic, message }) => {
        const event: CDCEvent = JSON.parse(message.value.toString());

        switch (event.op) {
          case 'c': // CREATE
            await this.handleInsert(event.after);
            break;
          case 'u': // UPDATE
            await this.handleUpdate(event.before, event.after);
            break;
          case 'd': // DELETE
            await this.handleDelete(event.before);
            break;
        }
      },
    });
  }

  private async handleInsert(record: any): Promise<void> {
    console.log('Insert:', record);
    // Sync to cache, search index, etc.
  }

  private async handleUpdate(before: any, after: any): Promise<void> {
    console.log('Update:', before, '->', after);
    // Invalidate cache, update search index
  }

  private async handleDelete(record: any): Promise<void> {
    console.log('Delete:', record);
    // Remove from cache, search index
  }
}

Python Consumer

from confluent_kafka import Consumer
import json

class CDCConsumer:
    def __init__(self, bootstrap_servers: str, group_id: str):
        self.consumer = Consumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id,
            'auto.offset.reset': 'earliest',
        })

    def subscribe(self, tables: list):
        topics = [f'mydb.inventory.{table}' for table in tables]
        self.consumer.subscribe(topics)

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

            event = json.loads(msg.value().decode('utf-8'))

            if event['op'] == 'c':
                self.handle_insert(event['after'])
            elif event['op'] == 'u':
                self.handle_update(event['before'], event['after'])
            elif event['op'] == 'd':
                self.handle_delete(event['before'])

    def handle_insert(self, record):
        print(f'Insert: {record}')
        # Sync to Elasticsearch, Redis, etc.

    def handle_update(self, before, after):
        print(f'Update: {before} -> {after}')

    def handle_delete(self, record):
        print(f'Delete: {record}')

# Usage
consumer = CDCConsumer('localhost:9092', 'cdc-consumer')
consumer.subscribe(['users', 'orders'])
consumer.consume()

Outbox Pattern

Combine CDC with outbox pattern for reliable event publishing:

1. Create Outbox Table

CREATE TABLE outbox (
  id UUID PRIMARY KEY,
  aggregate_id VARCHAR(255),
  event_type VARCHAR(255),
  payload JSONB,
  created_at TIMESTAMP DEFAULT NOW()
);

2. Transactional Write

BEGIN;
  -- Business logic
  UPDATE orders SET status = 'shipped' WHERE id = '123';

  -- Write to outbox
  INSERT INTO outbox (id, aggregate_id, event_type, payload)
  VALUES (
    gen_random_uuid(),
    '123',
    'OrderShipped',
    '{"orderId": "123", "trackingNumber": "TRACK123"}'::jsonb
  );
COMMIT;

3. CDC Captures Outbox

Debezium captures outbox changes and publishes to Kafka. Application consumes events from Kafka topic.

Best Practices

1. Use Debezium: Industry-standard, battle-tested 2. Monitor lag: Track replication delay 3. Handle schema changes: Plan for column additions/removals 4. Idempotent consumers: CDC may deliver duplicates 5. Filter events: Use SMTs (Single Message Transforms) 6. Tombstone events: Handle deletes properly

Monitoring

from prometheus_client import Gauge

cdc_lag = Gauge('cdc_replication_lag_seconds', 'CDC replication lag')

def monitor_lag(event):
    current_time = time.time() * 1000
    event_time = event['ts_ms']
    lag_ms = current_time - event_time
    cdc_lag.set(lag_ms / 1000)

Conclusion

CDC enables real-time data synchronization without application code changes. Use Debezium for production deployments, implement the outbox pattern for transactional guarantees.

Delivery Guarantees in Stream Processing

Overview
At-Most-Once Delivery
At-Least-Once Delivery
Exactly-Once Delivery
Comparison Matrix
Configuration Summary
Best Practices
Conclusion

Overview

Stream processing systems offer three delivery semantics: at-most-once, at-least-once, and exactly-once. Choose based on use case requirements and acceptable trade-offs.

At-Most-Once Delivery

Characteristics

Messages may be lost
No duplicates
Lowest overhead and complexity
Best performance

Implementation

Consumer commits offset before processing message
Producer sends without acknowledgement (acks=0)

Use Cases

Metrics and monitoring (loss acceptable)
Log aggregation (sampling OK)
Best-effort notifications

Example (Python)

# Consumer commits before processing
consumer = Consumer({
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'enable.auto.commit': True,  # Auto-commit before processing
    'auto.commit.interval.ms': 1000,
})

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

    # Message may be lost if processing fails
    try:
        process_message(msg.value())
    except Exception as e:
        # Offset already committed - message lost
        logging.error(f"Message lost: {e}")

At-Least-Once Delivery

Characteristics

Messages never lost (guaranteed delivery)
May have duplicates (redelivery on failure)
Moderate overhead
Most common choice for production systems

Implementation

Consumer commits offset after processing message
Producer waits for acknowledgement (acks=all)
Idempotent message processing required

Use Cases

Most production applications
Order processing (with idempotency)
Database synchronization
Event-driven architectures

Example (TypeScript)

// Producer: acks=all
const producer = kafka.producer({
  idempotent: true,  // Prevents duplicates from retries
});

await producer.send({
  topic: 'orders',
  acks: -1,  // Wait for all replicas
  messages: [{ value: JSON.stringify(order) }],
});

// Consumer: manual commit after processing
await consumer.run({
  autoCommit: false,
  eachMessage: async ({ topic, partition, message }) => {
    try {
      // Process message idempotently
      await processMessageIdempotently(message);

      // Commit offset only after successful processing
      await consumer.commitOffsets([{
        topic,
        partition,
        offset: (parseInt(message.offset) + 1).toString(),
      }]);
    } catch (error) {
      // Don't commit - message will be reprocessed
      console.error('Processing failed, will retry:', error);
    }
  },
});

Idempotency Strategies

1. Deduplication by Message ID

const processedIds = new Set<string>();

async function processMessageIdempotently(message: any) {
  const messageId = message.headers['message-id'];

  if (processedIds.has(messageId)) {
    console.log('Duplicate message, skipping');
    return;
  }

  await processMessage(message);
  processedIds.add(messageId);
}

2. Database Unique Constraints

CREATE TABLE orders (
  order_id VARCHAR(36) PRIMARY KEY,
  -- other fields
);

-- Insert will fail silently if duplicate
INSERT INTO orders (order_id, ...)
VALUES (?, ...)
ON DUPLICATE KEY UPDATE order_id = order_id;

3. Check-then-Set Pattern

async function processOrderIdempotently(order: Order) {
  const existing = await db.orders.findOne({ orderId: order.id });

  if (existing) {
    console.log('Order already processed');
    return;
  }

  await db.orders.insert(order);
}

Exactly-Once Delivery

Characteristics

Messages never lost and never duplicated
Highest overhead and complexity
Requires transactional support
End-to-end exactly-once (source to sink)

Implementation

Producer uses transactions
Consumer processes and commits offset in same transaction
Idempotent producers (enable.idempotence=true)

Use Cases

Financial transactions
Payment processing
Critical state updates
Compliance-sensitive data

Example (Java)

// Producer with transactions
Properties props = new Properties();
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-transactional-id");
props.put(ProducerConfig.ACKS_CONFIG, "all");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

// Initialize transactions
producer.initTransactions();

try {
    producer.beginTransaction();

    // Send messages
    producer.send(new ProducerRecord<>("topic1", "key", "value"));
    producer.send(new ProducerRecord<>("topic2", "key", "value"));

    // Commit transaction
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
    throw e;
}

Exactly-Once with Consumer + Producer

// Consumer with exactly-once processing
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);

Properties producerProps = new Properties();
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "consumer-producer-tx");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);
producer.initTransactions();

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

    if (!records.isEmpty()) {
        producer.beginTransaction();

        try {
            for (ConsumerRecord<String, String> record : records) {
                // Process message
                String result = processMessage(record.value());

                // Send output
                producer.send(new ProducerRecord<>("output", result));
            }

            // Commit consumer offsets in same transaction
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
            for (ConsumerRecord<String, String> record : records) {
                offsets.put(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
                );
            }
            producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());

            // Commit transaction
            producer.commitTransaction();
        } catch (Exception e) {
            producer.abortTransaction();
        }
    }
}

Comparison Matrix

Guarantee	Message Loss	Duplicates	Overhead	Use Case
At-Most-Once	Possible	No	Low	Metrics, logs
At-Least-Once	No	Possible	Medium	Most applications
Exactly-Once	No	No	High	Financial, critical

Configuration Summary

At-Most-Once Configuration

Producer:

acks=0                          # No acknowledgement
enable.idempotence=false

Consumer:

enable.auto.commit=true         # Commit before processing
auto.commit.interval.ms=1000

At-Least-Once Configuration

Producer:

acks=all                        # Wait for all replicas
retries=Integer.MAX_VALUE
enable.idempotence=true         # Prevent duplicates from retries
max.in.flight.requests.per.connection=5

Consumer:

enable.auto.commit=false        # Manual commit after processing

Exactly-Once Configuration

Producer:

enable.idempotence=true
transactional.id=unique-tx-id
acks=all
max.in.flight.requests.per.connection=5

Consumer:

enable.auto.commit=false
isolation.level=read_committed  # Read only committed transactions

Best Practices

1. Start with At-Least-Once

Most applications should use at-least-once delivery with idempotent processing. It provides good reliability without the complexity of exactly-once.

2. Design for Idempotency

Even with at-least-once, design message processing to be idempotent:

Use unique message IDs
Leverage database constraints
Implement check-then-set patterns

3. Use Exactly-Once Sparingly

Only use exactly-once when absolutely required (financial transactions, compliance). The added complexity and overhead are significant.

4. Monitor Delivery Metrics

Track metrics for:

Message loss (at-most-once)
Duplicate processing rate (at-least-once)
Transaction abort rate (exactly-once)

5. Test Failure Scenarios

Test behavior under:

Network partitions
Consumer crashes
Broker failures
Slow processing

Conclusion

Default recommendation: Use at-least-once delivery with idempotent message processing for most applications. Reserve exactly-once for critical use cases where duplicates are unacceptable.

Error Handling Patterns for Stream Processing

Overview
Dead Letter Queue (DLQ) Pattern
Retry Strategies
Backpressure Handling
Circuit Breaker Pattern
Graceful Shutdown
Monitoring and Alerting
Best Practices
Conclusion

Overview

Production stream processing requires robust error handling strategies including dead-letter queues, retry logic, backpressure management, and circuit breakers.

Dead Letter Queue (DLQ) Pattern

Purpose

Send messages that fail processing to a separate topic for later analysis and manual intervention.

When to Use

Message cannot be parsed (schema mismatch)
Processing fails after maximum retries
Downstream service permanently unavailable
Business logic validation failures

Implementation (TypeScript)

class ConsumerWithDLQ {
  private consumer: Consumer;
  private dlqProducer: Producer;

  async processWithDLQ(message: any): Promise<void> {
    try {
      await this.processMessage(message);
      await this.consumer.commitOffsets([...]);
    } catch (error) {
      await this.sendToDLQ(message, error);
      await this.consumer.commitOffsets([...]);
    }
  }

  private async sendToDLQ(message: any, error: Error): Promise<void> {
    const dlqTopic = `${message.topic}.dlq`;

    await this.dlqProducer.send({
      topic: dlqTopic,
      messages: [{
        key: message.key,
        value: message.value,
        headers: {
          'original-topic': message.topic,
          'error-message': error.message,
          'error-stack': error.stack,
          'failed-at': new Date().toISOString(),
          'retry-count': '3',
        },
      }],
    });
  }
}

Retry Strategies

Exponential Backoff

import time

def process_with_retry(message, max_retries=3):
    for attempt in range(max_retries):
        try:
            process_message(message)
            return True
        except Exception as e:
            if attempt < max_retries - 1:
                # Exponential backoff: 1s, 2s, 4s
                sleep_time = 2 ** attempt
                time.sleep(sleep_time)
            else:
                # Final attempt failed
                send_to_dlq(message, e)
                return False

Retry with Jitter

func processWithRetry(msg kafka.Message, maxRetries int) error {
    for attempt := 0; attempt < maxRetries; attempt++ {
        err := processMessage(msg)
        if err == nil {
            return nil
        }

        if attempt < maxRetries-1 {
            // Exponential backoff with jitter
            baseDelay := time.Duration(math.Pow(2, float64(attempt))) * time.Second
            jitter := time.Duration(rand.Int63n(int64(baseDelay / 2)))
            time.Sleep(baseDelay + jitter)
        }
    }

    return sendToDLQ(msg)
}

Backpressure Handling

Consumer Backpressure

public class BackpressureConsumer {
    private final Semaphore semaphore;
    private final int maxConcurrent = 100;

    public BackpressureConsumer() {
        this.semaphore = new Semaphore(maxConcurrent);
    }

    public void consume() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

            for (ConsumerRecord<String, String> record : records) {
                try {
                    // Acquire permit (blocks if maxConcurrent reached)
                    semaphore.acquire();

                    // Process asynchronously
                    executor.submit(() -> {
                        try {
                            processMessage(record);
                        } finally {
                            semaphore.release();
                        }
                    });
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }
}

Producer Backpressure

class BackpressureProducer {
  private queue: any[] = [];
  private maxQueueSize = 10000;
  private processing = false;

  async send(message: any): Promise<void> {
    if (this.queue.length >= this.maxQueueSize) {
      throw new Error('Queue full - backpressure applied');
    }

    this.queue.push(message);

    if (!this.processing) {
      this.processBatch();
    }
  }

  private async processBatch(): Promise<void> {
    this.processing = true;

    while (this.queue.length > 0) {
      const batch = this.queue.splice(0, 100);

      try {
        await this.producer.sendBatch(batch);
      } catch (error) {
        // Re-queue on failure
        this.queue.unshift(...batch);
        await sleep(1000);
      }
    }

    this.processing = false;
  }
}

Circuit Breaker Pattern

enum CircuitState {
  CLOSED,
  OPEN,
  HALF_OPEN,
}

class CircuitBreaker {
  private state = CircuitState.CLOSED;
  private failureCount = 0;
  private successCount = 0;
  private lastFailureTime = 0;

  constructor(
    private threshold = 5,
    private timeout = 60000,
    private halfOpenAttempts = 3
  ) {}

  async execute<T>(fn: () => Promise<T>): Promise<T> {
    if (this.state === CircuitState.OPEN) {
      if (Date.now() - this.lastFailureTime > this.timeout) {
        this.state = CircuitState.HALF_OPEN;
        this.successCount = 0;
      } else {
        throw new Error('Circuit breaker is OPEN');
      }
    }

    try {
      const result = await fn();
      this.onSuccess();
      return result;
    } catch (error) {
      this.onFailure();
      throw error;
    }
  }

  private onSuccess(): void {
    this.failureCount = 0;

    if (this.state === CircuitState.HALF_OPEN) {
      this.successCount++;
      if (this.successCount >= this.halfOpenAttempts) {
        this.state = CircuitState.CLOSED;
      }
    }
  }

  private onFailure(): void {
    this.failureCount++;
    this.lastFailureTime = Date.now();

    if (this.failureCount >= this.threshold) {
      this.state = CircuitState.OPEN;
    }
  }
}

Graceful Shutdown

TypeScript

class GracefulConsumer {
  private isShuttingDown = false;

  async start(): Promise<void> {
    process.on('SIGTERM', () => this.shutdown());
    process.on('SIGINT', () => this.shutdown());

    await this.consume();
  }

  private async shutdown(): Promise<void> {
    if (this.isShuttingDown) return;

    console.log('Shutting down gracefully...');
    this.isShuttingDown = true;

    // Stop accepting new messages
    await this.consumer.disconnect();

    // Wait for in-flight messages to complete
    await this.waitForInFlight();

    process.exit(0);
  }
}

Go

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    go func() {
        <-sigChan
        log.Println("Shutdown signal received")
        cancel()
    }()

    consumer.Consume(ctx, handler)
    log.Println("Shutdown complete")
}

Monitoring and Alerting

Metrics to Track

from prometheus_client import Counter, Histogram, Gauge

# Message metrics
messages_processed = Counter('messages_processed_total', 'Total messages processed')
messages_failed = Counter('messages_failed_total', 'Total messages failed')
processing_duration = Histogram('processing_duration_seconds', 'Message processing duration')

# Consumer lag
consumer_lag = Gauge('consumer_lag', 'Consumer lag in messages', ['partition'])

# DLQ metrics
dlq_messages = Counter('dlq_messages_total', 'Messages sent to DLQ')

def process_message(msg):
    start_time = time.time()

    try:
        # Process message
        handle_message(msg)

        messages_processed.inc()
    except Exception as e:
        messages_failed.inc()
        send_to_dlq(msg, e)
        dlq_messages.inc()
    finally:
        duration = time.time() - start_time
        processing_duration.observe(duration)

Best Practices

1. Always implement DLQ: Never silently drop failed messages 2. Use exponential backoff: Avoid thundering herd on retry 3. Set max retries: Prevent infinite retry loops 4. Monitor consumer lag: Alert on growing backlog 5. Graceful shutdown: Finish processing before exit 6. Circuit breakers: Protect downstream services 7. Backpressure: Prevent memory exhaustion

Conclusion

Robust error handling is critical for production stream processing. Implement DLQ patterns, retry logic with backoff, backpressure management, and graceful shutdown to build reliable systems.

Event Sourcing Patterns

Overview
Core Concepts
Benefits
Challenges
Implementation Pattern
Event Schema Evolution
Snapshots
Best Practices
Conclusion

Overview

Event sourcing stores all changes to application state as a sequence of immutable events. Instead of storing current state, store the history of state changes.

Core Concepts

Event Store

Append-only log of all events (Kafka is ideal for this)

Event

Immutable fact that something happened

Aggregate

Entity whose state is derived from events

Projection

Read model built from event stream

Benefits

Complete audit trail
Temporal queries (state at any point in time)
Event replay for debugging
Easy to add new projections
Natural fit for event-driven architecture

Challenges

Event schema evolution
Eventual consistency
Increased storage requirements
Complexity in querying current state

Implementation Pattern

Define Events

interface Event {
  eventId: string;
  eventType: string;
  aggregateId: string;
  timestamp: number;
  version: number;
  data: any;
}

interface OrderCreatedEvent extends Event {
  eventType: 'OrderCreated';
  data: {
    orderId: string;
    customerId: string;
    items: OrderItem[];
    total: number;
  };
}

interface OrderShippedEvent extends Event {
  eventType: 'OrderShipped';
  data: {
    orderId: string;
    trackingNumber: string;
    shippedAt: number;
  };
}

Event Store (Kafka)

class EventStore {
  private producer: Producer;
  private consumer: Consumer;

  async appendEvent(event: Event): Promise<void> {
    await this.producer.send({
      topic: 'events',
      messages: [{
        key: event.aggregateId,
        value: JSON.stringify(event),
        headers: {
          'event-type': event.eventType,
          'event-version': event.version.toString(),
        },
      }],
    });
  }

  async getEvents(aggregateId: string): Promise<Event[]> {
    // Read all events for aggregate from beginning
    const events: Event[] = [];

    await this.consumer.subscribe({
      topics: ['events'],
      fromBeginning: true,
    });

    await this.consumer.run({
      eachMessage: async ({ message }) => {
        const event = JSON.parse(message.value.toString());
        if (event.aggregateId === aggregateId) {
          events.push(event);
        }
      },
    });

    return events;
  }
}

Aggregate

class Order {
  private id: string;
  private customerId: string;
  private items: OrderItem[] = [];
  private status: OrderStatus = 'pending';
  private version = 0;
  private uncommittedEvents: Event[] = [];

  static async load(id: string, eventStore: EventStore): Promise<Order> {
    const events = await eventStore.getEvents(id);
    const order = new Order(id);

    for (const event of events) {
      order.applyEvent(event, false);
    }

    return order;
  }

  createOrder(customerId: string, items: OrderItem[]): void {
    const event: OrderCreatedEvent = {
      eventId: uuid(),
      eventType: 'OrderCreated',
      aggregateId: this.id,
      timestamp: Date.now(),
      version: ++this.version,
      data: { orderId: this.id, customerId, items, total: calculateTotal(items) },
    };

    this.applyEvent(event, true);
  }

  shipOrder(trackingNumber: string): void {
    if (this.status !== 'pending') {
      throw new Error('Order already shipped');
    }

    const event: OrderShippedEvent = {
      eventId: uuid(),
      eventType: 'OrderShipped',
      aggregateId: this.id,
      timestamp: Date.now(),
      version: ++this.version,
      data: { orderId: this.id, trackingNumber, shippedAt: Date.now() },
    };

    this.applyEvent(event, true);
  }

  private applyEvent(event: Event, isNew: boolean): void {
    switch (event.eventType) {
      case 'OrderCreated':
        this.customerId = event.data.customerId;
        this.items = event.data.items;
        break;
      case 'OrderShipped':
        this.status = 'shipped';
        break;
    }

    if (isNew) {
      this.uncommittedEvents.push(event);
    }
  }

  async save(eventStore: EventStore): Promise<void> {
    for (const event of this.uncommittedEvents) {
      await eventStore.appendEvent(event);
    }
    this.uncommittedEvents = [];
  }
}

Projection (Read Model)

class OrderProjection {
  private db: Database;

  async build(eventStream: EventStream): Promise<void> {
    await eventStream.subscribe(['events'], async (event: Event) => {
      switch (event.eventType) {
        case 'OrderCreated':
          await this.db.orders.insert({
            id: event.data.orderId,
            customerId: event.data.customerId,
            total: event.data.total,
            status: 'pending',
          });
          break;

        case 'OrderShipped':
          await this.db.orders.update(
            { id: event.data.orderId },
            { status: 'shipped', trackingNumber: event.data.trackingNumber }
          );
          break;
      }
    });
  }

  async getOrder(orderId: string): Promise<Order> {
    return await this.db.orders.findOne({ id: orderId });
  }
}

Event Schema Evolution

Versioning Strategy

interface EventV1 {
  version: 1;
  eventType: 'OrderCreated';
  data: {
    orderId: string;
    customerId: string;
  };
}

interface EventV2 {
  version: 2;
  eventType: 'OrderCreated';
  data: {
    orderId: string;
    customerId: string;
    customerEmail: string; // New field
  };
}

function migrateEvent(event: Event): EventV2 {
  if (event.version === 1) {
    return {
      ...event,
      version: 2,
      data: {
        ...event.data,
        customerEmail: 'unknown@example.com', // Default value
      },
    };
  }
  return event;
}

Snapshots

For aggregates with many events, use snapshots:

class SnapshotStore {
  async saveSnapshot(aggregateId: string, state: any, version: number): Promise<void> {
    await this.db.snapshots.upsert({
      aggregateId,
      state: JSON.stringify(state),
      version,
      createdAt: Date.now(),
    });
  }

  async getSnapshot(aggregateId: string): Promise<Snapshot | null> {
    return await this.db.snapshots.findOne({ aggregateId });
  }
}

class Order {
  static async load(id: string, eventStore: EventStore, snapshotStore: SnapshotStore): Promise<Order> {
    const snapshot = await snapshotStore.getSnapshot(id);

    let order: Order;
    let fromVersion = 0;

    if (snapshot) {
      order = Order.fromSnapshot(snapshot);
      fromVersion = snapshot.version;
    } else {
      order = new Order(id);
    }

    // Load events after snapshot
    const events = await eventStore.getEvents(id, fromVersion);
    for (const event of events) {
      order.applyEvent(event, false);
    }

    return order;
  }
}

Best Practices

1. Immutable events: Never modify published events 2. Idempotent projections: Handle duplicate events 3. Event versioning: Plan for schema evolution 4. Snapshots: Use for aggregates with many events 5. Upcasting: Convert old events to new schema 6. Correlation IDs: Track causation across events

Conclusion

Event sourcing provides a complete audit trail and enables temporal queries. Use Kafka as the event store, design for schema evolution, and implement snapshots for performance.

Exactly-Once Processing

Overview
Requirements
Idempotent Producer
Transactions
TypeScript Transactions (KafkaJS)
Performance Considerations
When to Use
Alternatives to Transactions
Monitoring
Best Practices
Conclusion

Overview

Exactly-once semantics guarantee that messages are processed exactly once - neither lost nor duplicated. Critical for financial transactions and stateful processing.

Requirements

1. Idempotent producers: Prevent duplicate writes from retries 2. Transactions: Atomic writes to multiple topics 3. Transactional reads: Consumers read only committed data 4. Offset commits in transactions: Atomic processing + commit

Idempotent Producer

Configuration

enable.idempotence=true
acks=all
retries=Integer.MAX_VALUE
max.in.flight.requests.per.connection=5

How It Works

Producer assigns sequence numbers to messages. Broker detects and deduplicates based on producer ID + sequence number.

TypeScript

const producer = kafka.producer({
  idempotent: true,
  maxInFlightRequests: 5,
});

Transactions

Producer-Only Transactions

Write to multiple topics atomically:

Properties props = new Properties();
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "my-tx-id");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();

    // Send to multiple topics
    producer.send(new ProducerRecord<>("topic1", "key", "value1"));
    producer.send(new ProducerRecord<>("topic2", "key", "value2"));

    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
    throw e;
}

Consumer-Producer Transactions

Process message and produce output atomically:

Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

Properties producerProps = new Properties();
producerProps.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "consumer-producer-tx");
producerProps.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
KafkaProducer<String, String> producer = new KafkaProducer<>(producerProps);

producer.initTransactions();
consumer.subscribe(Arrays.asList("input-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

    if (!records.isEmpty()) {
        producer.beginTransaction();

        try {
            // Process records
            for (ConsumerRecord<String, String> record : records) {
                String result = processMessage(record.value());

                // Send output
                producer.send(new ProducerRecord<>("output-topic", result));
            }

            // Commit consumer offsets in transaction
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
            for (ConsumerRecord<String, String> record : records) {
                offsets.put(
                    new TopicPartition(record.topic(), record.partition()),
                    new OffsetAndMetadata(record.offset() + 1)
                );
            }

            producer.sendOffsetsToTransaction(offsets, consumer.groupMetadata());

            // Commit transaction
            producer.commitTransaction();

        } catch (Exception e) {
            producer.abortTransaction();
        }
    }
}

TypeScript Transactions (KafkaJS)

import { Kafka, Producer, Consumer } from 'kafkajs';

class ExactlyOnceProcessor {
  private producer: Producer;
  private consumer: Consumer;

  constructor(kafka: Kafka, transactionalId: string) {
    this.producer = kafka.producer({
      idempotent: true,
      maxInFlightRequests: 1,
      transactionalId,
    });

    this.consumer = kafka.consumer({
      groupId: 'my-group',
      readUncommitted: false, // Read only committed messages
    });
  }

  async process(): Promise<void> {
    await this.producer.connect();
    await this.consumer.connect();
    await this.consumer.subscribe({ topics: ['input'] });

    await this.consumer.run({
      autoCommit: false,
      eachMessage: async ({ topic, partition, message }) => {
        const transaction = await this.producer.transaction();

        try {
          const value = message.value?.toString();
          const result = await this.processMessage(value);

          // Send output in transaction
          await transaction.send({
            topic: 'output',
            messages: [{ value: result }],
          });

          // Commit consumer offset in same transaction
          await transaction.sendOffsets({
            consumerGroupId: this.consumer.groupId,
            topics: [{
              topic,
              partitions: [{
                partition,
                offset: (parseInt(message.offset) + 1).toString(),
              }],
            }],
          });

          await transaction.commit();

        } catch (error) {
          await transaction.abort();
          throw error;
        }
      },
    });
  }

  private async processMessage(input: string): Promise<string> {
    // Business logic
    return input.toUpperCase();
  }
}

Performance Considerations

Overhead

Exactly-once semantics add overhead:

Latency: +20-50% vs at-least-once
Throughput: -10-30% vs at-least-once
Resource usage: Higher memory, CPU

Tuning

# Reduce latency
transaction.timeout.ms=60000
transaction.state.log.min.isr=1

# Increase throughput
transaction.state.log.replication.factor=3

When to Use

Use exactly-once when:

Financial transactions
Payment processing
Inventory management
Compliance requirements
Stateful aggregations

Avoid exactly-once when:

Metrics, logs (loss acceptable)
High-volume, low-value events
Performance is critical
Idempotent consumers sufficient

Alternatives to Transactions

1. Idempotent Processing

Design consumers to handle duplicates:

processed_ids = set()

def process_idempotent(message):
    message_id = message.headers['message-id']

    if message_id in processed_ids:
        print('Duplicate, skipping')
        return

    process_message(message)
    processed_ids.add(message_id)

2. Database Constraints

Use unique constraints:

INSERT INTO orders (order_id, ...)
VALUES (?, ...)
ON DUPLICATE KEY UPDATE order_id = order_id;

3. External Transaction Coordinator

Use distributed transaction frameworks:

Apache Flink state backend
Apache Spark checkpointing
Custom coordination service

Monitoring

from prometheus_client import Counter, Histogram

tx_commits = Counter('kafka_tx_commits_total', 'Transactions committed')
tx_aborts = Counter('kafka_tx_aborts_total', 'Transactions aborted')
tx_duration = Histogram('kafka_tx_duration_seconds', 'Transaction duration')

def process_with_tx():
    start = time.time()

    try:
        transaction.begin()
        # Processing logic
        transaction.commit()
        tx_commits.inc()
    except Exception:
        transaction.abort()
        tx_aborts.inc()
        raise
    finally:
        tx_duration.observe(time.time() - start)

Best Practices

1. Use unique transactional IDs: One per producer instance 2. Set appropriate timeouts: Balance reliability vs performance 3. Monitor abort rate: High aborts indicate issues 4. Test failure scenarios: Crashes, network partitions 5. Consider alternatives: Idempotent processing often sufficient

Conclusion

Exactly-once semantics provide strong guarantees but add complexity and overhead. Use for critical use cases like financial transactions. For most applications, at-least-once with idempotent processing is sufficient.

Go Streaming Patterns (kafka-go)

Overview
Installation
Basic Producer
Basic Consumer
Concurrent Consumer
Graceful Shutdown
Best Practices
Conclusion

Overview

kafka-go (Segment) provides an idiomatic Go API for Kafka with zero external dependencies. It mirrors Go's standard library design patterns.

Library: kafka-go (segmentio/kafka-go) Code Snippets: 42+ Trust Score: High Best For: High-performance microservices, infrastructure tools

Installation

go get github.com/segmentio/kafka-go

Basic Producer

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "time"

    "github.com/segmentio/kafka-go"
)

type Event struct {
    UserID    string `json:"user_id"`
    Action    string `json:"action"`
    Timestamp int64  `json:"timestamp"`
}

type EventProducer struct {
    writer *kafka.Writer
}

func NewEventProducer(brokers []string, topic string) *EventProducer {
    return &EventProducer{
        writer: &kafka.Writer{
            Addr:     kafka.TCP(brokers...),
            Topic:    topic,
            Balancer: &kafka.LeastBytes{},
            // At-least-once semantics
            RequiredAcks: kafka.RequireAll,
            MaxAttempts:  3,
            // Performance
            Compression: kafka.Gzip,
            BatchSize:   100,
            BatchTimeout: 10 * time.Millisecond,
        },
    }
}

func (p *EventProducer) SendEvent(ctx context.Context, event Event) error {
    value, err := json.Marshal(event)
    if err != nil {
        return fmt.Errorf("marshal event: %w", err)
    }

    err = p.writer.WriteMessages(ctx, kafka.Message{
        Key:   []byte(event.UserID),
        Value: value,
        Headers: []kafka.Header{
            {Key: "event-type", Value: []byte(event.Action)},
        },
    })

    if err != nil {
        return fmt.Errorf("write message: %w", err)
    }

    return nil
}

func (p *EventProducer) Close() error {
    return p.writer.Close()
}

// Usage
func main() {
    producer := NewEventProducer([]string{"localhost:9092"}, "user-actions")
    defer producer.Close()

    event := Event{
        UserID:    "user-123",
        Action:    "login",
        Timestamp: time.Now().Unix(),
    }

    if err := producer.SendEvent(context.Background(), event); err != nil {
        fmt.Printf("Failed: %v\n", err)
    }
}

Basic Consumer

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"

    "github.com/segmentio/kafka-go"
)

type EventConsumer struct {
    reader *kafka.Reader
}

func NewEventConsumer(brokers []string, topic string, groupID string) *EventConsumer {
    return &EventConsumer{
        reader: kafka.NewReader(kafka.ReaderConfig{
            Brokers:  brokers,
            Topic:    topic,
            GroupID:  groupID,
            MaxBytes: 10e6, // 10MB
            // Manual commit
            CommitInterval: 0,
        }),
    }
}

func (c *EventConsumer) Consume(ctx context.Context, handler func(Event) error) error {
    for {
        msg, err := c.reader.FetchMessage(ctx)
        if err != nil {
            if err == context.Canceled {
                break
            }
            log.Printf("Fetch error: %v", err)
            continue
        }

        var event Event
        if err := json.Unmarshal(msg.Value, &event); err != nil {
            log.Printf("Unmarshal error: %v", err)
            // Send to DLQ
            c.sendToDLQ(msg)
            c.reader.CommitMessages(ctx, msg)
            continue
        }

        // Process message
        if err := handler(event); err != nil {
            log.Printf("Handler error: %v", err)
            // Don't commit - will be reprocessed
            continue
        }

        // Commit after successful processing
        if err := c.reader.CommitMessages(ctx, msg); err != nil {
            log.Printf("Commit error: %v", err)
        }
    }

    return nil
}

func (c *EventConsumer) sendToDLQ(msg kafka.Message) {
    // DLQ implementation
}

func (c *EventConsumer) Close() error {
    return c.reader.Close()
}

// Usage
func main() {
    consumer := NewEventConsumer(
        []string{"localhost:9092"},
        "user-actions",
        "my-consumer-group",
    )
    defer consumer.Close()

    handler := func(event Event) error {
        fmt.Printf("Processing: %+v\n", event)
        return nil
    }

    if err := consumer.Consume(context.Background(), handler); err != nil {
        log.Fatal(err)
    }
}

Concurrent Consumer

type ConcurrentConsumer struct {
    reader  *kafka.Reader
    workers int
}

func NewConcurrentConsumer(brokers []string, topic, groupID string, workers int) *ConcurrentConsumer {
    return &ConcurrentConsumer{
        reader: kafka.NewReader(kafka.ReaderConfig{
            Brokers: brokers,
            Topic:   topic,
            GroupID: groupID,
        }),
        workers: workers,
    }
}

func (c *ConcurrentConsumer) Consume(ctx context.Context, handler func(Event) error) error {
    messageChan := make(chan kafka.Message, c.workers*2)
    errorChan := make(chan error, c.workers)

    // Start worker pool
    for i := 0; i < c.workers; i++ {
        go c.worker(ctx, messageChan, errorChan, handler)
    }

    // Fetch messages
    for {
        msg, err := c.reader.FetchMessage(ctx)
        if err != nil {
            if err == context.Canceled {
                break
            }
            log.Printf("Fetch error: %v", err)
            continue
        }

        select {
        case messageChan <- msg:
        case err := <-errorChan:
            log.Printf("Worker error: %v", err)
        case <-ctx.Done():
            close(messageChan)
            return ctx.Err()
        }
    }

    close(messageChan)
    return nil
}

func (c *ConcurrentConsumer) worker(
    ctx context.Context,
    messages <-chan kafka.Message,
    errors chan<- error,
    handler func(Event) error,
) {
    for msg := range messages {
        var event Event
        if err := json.Unmarshal(msg.Value, &event); err != nil {
            errors <- err
            c.reader.CommitMessages(ctx, msg)
            continue
        }

        if err := handler(event); err != nil {
            errors <- err
            continue
        }

        c.reader.CommitMessages(ctx, msg)
    }
}

Graceful Shutdown

package main

import (
    "context"
    "os"
    "os/signal"
    "syscall"
)

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Handle signals
    sigChan := make(chan os.Signal, 1)
    signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)

    consumer := NewEventConsumer(
        []string{"localhost:9092"},
        "events",
        "my-group",
    )
    defer consumer.Close()

    go func() {
        <-sigChan
        fmt.Println("Shutting down...")
        cancel()
    }()

    handler := func(event Event) error {
        fmt.Printf("Processing: %+v\n", event)
        return nil
    }

    if err := consumer.Consume(ctx, handler); err != nil && err != context.Canceled {
        fmt.Printf("Error: %v\n", err)
    }

    fmt.Println("Shutdown complete")
}

Best Practices

1. Use Context for Cancellation

ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()

err := producer.SendEvent(ctx, event)

2. Handle Errors Explicitly

if err := writer.WriteMessages(ctx, messages...); err != nil {
    if errors.Is(err, context.DeadlineExceeded) {
        // Timeout handling
    } else if errors.Is(err, kafka.LeaderNotAvailable) {
        // Retry
    } else {
        // Fatal error
    }
}

3. Batch Messages for Performance

messages := make([]kafka.Message, 0, 100)
for _, event := range events {
    value, _ := json.Marshal(event)
    messages = append(messages, kafka.Message{Value: value})
}

writer.WriteMessages(ctx, messages...)

Conclusion

kafka-go provides a performant, idiomatic Go client for Kafka. Use it for high-throughput microservices and infrastructure tools.

Java Streaming Patterns (Apache Kafka Java Client)

Overview
Installation
Basic Producer
Basic Consumer
Transactional Producer (Exactly-Once)
Spring Kafka Integration
Best Practices
Conclusion

Overview

The Apache Kafka Java Client is the most feature-complete Kafka client, offering full support for exactly-once semantics, transactions, and all advanced features.

Library: Apache Kafka Java Client Code Snippets: 683+ Trust Score: High (76.9) Best For: Enterprise applications, Kafka Streams, Flink, Spark

Installation

Maven:

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>3.6.0</version>
</dependency>

Gradle:

implementation 'org.apache.kafka:kafka-clients:3.6.0'

Basic Producer

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.Properties;
import java.util.concurrent.Future;

public class EventProducer {
    private final KafkaProducer<String, String> producer;
    private final ObjectMapper objectMapper;
    private final String topic;

    public EventProducer(String bootstrapServers, String topic) {
        this.topic = topic;
        this.objectMapper = new ObjectMapper();

        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // At-least-once delivery
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);

        // Performance tuning
        props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);

        this.producer = new KafkaProducer<>(props);
    }

    public void sendAsync(Event event) throws Exception {
        String key = event.getUserId();
        String value = objectMapper.writeValueAsString(event);

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    System.err.println("Send failed: " + exception.getMessage());
                } else {
                    System.out.printf("Sent to partition %d offset %d%n",
                        metadata.partition(), metadata.offset());
                }
            }
        });
    }

    public RecordMetadata sendSync(Event event) throws Exception {
        String key = event.getUserId();
        String value = objectMapper.writeValueAsString(event);

        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        Future<RecordMetadata> future = producer.send(record);
        return future.get(); // Block until complete
    }

    public void close() {
        producer.flush();
        producer.close();
    }
}

class Event {
    private String userId;
    private String action;
    private long timestamp;

    // Constructors, getters, setters
}

Basic Consumer

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.serialization.StringDeserializer;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class EventConsumer {
    private final KafkaConsumer<String, String> consumer;
    private final ObjectMapper objectMapper;

    public EventConsumer(String bootstrapServers, String groupId, String topic) {
        this.objectMapper = new ObjectMapper();

        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // Manual offset management
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        this.consumer = new KafkaConsumer<>(props);
        this.consumer.subscribe(Collections.singletonList(topic));
    }

    public void consume(EventHandler handler) {
        try {
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));

                for (ConsumerRecord<String, String> record : records) {
                    try {
                        Event event = objectMapper.readValue(record.value(), Event.class);

                        // Process event
                        handler.handle(event);

                        // Commit offset after successful processing
                        consumer.commitSync();

                    } catch (Exception e) {
                        System.err.println("Processing error: " + e.getMessage());
                        sendToDLQ(record);
                        consumer.commitSync();
                    }
                }
            }
        } finally {
            consumer.close();
        }
    }

    private void sendToDLQ(ConsumerRecord<String, String> record) {
        // DLQ implementation
    }
}

interface EventHandler {
    void handle(Event event) throws Exception;
}

Transactional Producer (Exactly-Once)

import org.apache.kafka.clients.producer.*;

public class TransactionalProducer {
    private final KafkaProducer<String, String> producer;

    public TransactionalProducer(String bootstrapServers, String transactionalId) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // Exactly-once configuration
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId);
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
        props.put(ProducerConfig.ACKS_CONFIG, "all");

        this.producer = new KafkaProducer<>(props);
        this.producer.initTransactions();
    }

    public void sendInTransaction(String topic, String key, String value) {
        producer.beginTransaction();

        try {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);
            producer.send(record);

            // Can send to multiple topics in same transaction
            producer.send(new ProducerRecord<>("audit-log", key, "processed"));

            producer.commitTransaction();
        } catch (Exception e) {
            producer.abortTransaction();
            throw e;
        }
    }

    public void close() {
        producer.close();
    }
}

Spring Kafka Integration

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

@Service
public class KafkaService {
    private final KafkaTemplate<String, Event> kafkaTemplate;

    public KafkaService(KafkaTemplate<String, Event> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void sendEvent(String topic, Event event) {
        kafkaTemplate.send(topic, event.getUserId(), event)
            .addCallback(
                result -> System.out.println("Sent: " + event),
                ex -> System.err.println("Failed: " + ex)
            );
    }

    @KafkaListener(topics = "user-actions", groupId = "my-group")
    public void listen(Event event) {
        System.out.println("Received: " + event);
        // Process event
    }
}

Best Practices

1. Use Try-With-Resources

try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
    producer.send(record).get();
}

2. Batch Processing

List<ProducerRecord<String, String>> batch = new ArrayList<>();
for (Event event : events) {
    batch.add(new ProducerRecord<>(topic, serialize(event)));
}

for (ProducerRecord<String, String> record : batch) {
    producer.send(record);
}
producer.flush();

3. Error Handling

try {
    producer.send(record).get();
} catch (ExecutionException e) {
    if (e.getCause() instanceof RetriableException) {
        // Retry
    } else {
        // Fatal error
    }
}

Conclusion

The Apache Kafka Java Client provides the most complete feature set for Kafka, including full transaction support for exactly-once semantics. Essential for enterprise Java applications.

Stream Processing Performance Tuning

Producer Performance
Consumer Performance
Kafka Broker Tuning
Partitioning Strategy
Monitoring Metrics
Conclusion

Producer Performance

Batching

# Increase batch size for throughput
batch.size=32768              # 32KB (default: 16KB)
linger.ms=10                  # Wait 10ms for batch
buffer.memory=67108864        # 64MB buffer

Compression

# LZ4: Best balance of speed/compression
compression.type=lz4

# GZIP: Better compression, slower
# compression.type=gzip

# Snappy: Fast, moderate compression
# compression.type=snappy

Partitioning

// Custom partitioner for even distribution
class CustomPartitioner {
  partition(topic: string, key: any, partitions: number): number {
    // Hash key to partition
    const hash = murmurhash(key);
    return hash % partitions;
  }
}

Consumer Performance

Parallelism

Match consumer instances to partition count:

Partitions: 10
Consumers: 10 (optimal)
Consumers: 5  (under-utilized)
Consumers: 15 (5 idle)

Fetch Settings

# Increase fetch size for throughput
fetch.min.bytes=10240         # 10KB
fetch.max.wait.ms=500         # Wait 500ms
max.partition.fetch.bytes=1048576  # 1MB per partition

Commit Frequency

# Less frequent commits = better performance
auto.commit.interval.ms=5000  # Commit every 5 seconds

Kafka Broker Tuning

Replica Settings

# Replication for fault tolerance
default.replication.factor=3
min.insync.replicas=2

Log Settings

# Segment size and retention
log.segment.bytes=1073741824  # 1GB segments
log.retention.hours=168       # 7 days
log.retention.bytes=-1        # No size limit

Network Threads

num.network.threads=8
num.io.threads=8
socket.send.buffer.bytes=1048576
socket.receive.buffer.bytes=1048576

Partitioning Strategy

Rule of Thumb

Partitions = (Target Throughput / Consumer Throughput)

Example:
Target: 1M msg/s
Consumer: 50k msg/s
Partitions = 1M / 50k = 20

Repartitioning

Too few partitions → add partitions (can't reduce):

kafka-topics.sh --alter --topic my-topic --partitions 20 \
  --bootstrap-server localhost:9092

Monitoring Metrics

Producer Metrics

record-send-rate: Messages/sec
record-error-rate: Errors/sec
request-latency-avg: Avg latency
batch-size-avg: Avg batch size

Consumer Metrics

records-consumed-rate: Messages/sec
fetch-latency-avg: Fetch latency
records-lag-max: Max lag
commit-latency-avg: Commit latency

Broker Metrics

BytesInPerSec: Inbound throughput
BytesOutPerSec: Outbound throughput
MessagesInPerSec: Message rate
UnderReplicatedPartitions: Replication lag

Conclusion

Tune batch size, compression, partitioning, and parallelism for optimal performance. Monitor key metrics and adjust based on workload patterns.

Stream Processor Selection Guide

Overview
Apache Flink
Apache Spark Streaming
Kafka Streams
ksqlDB
Comparison Matrix
Selection Flowchart

Overview

Stream processors transform, aggregate, and analyze streaming data in real-time. Choose based on latency requirements, processing model, deployment constraints, and ecosystem integration.

Apache Flink

Architecture

True stream processing (event-by-event)
Distributed dataflow engine
Stateful operators with checkpointing
Event-time processing with watermarks

Strengths

Millisecond-level latency
Superior state management (RocksDB backend)
Event-time semantics (handles out-of-order events)
Exactly-once processing guarantees
Complex Event Processing (CEP) support
Flexible windowing (tumbling, sliding, session)

Weaknesses

Steeper learning curve
Requires separate cluster deployment
Limited Python support (PyFlink improving but not mature)
Smaller ecosystem than Spark

Best Use Cases

Real-time analytics dashboards
Fraud detection (sub-second response)
Monitoring and alerting systems
Complex Event Processing (CEP)
High-frequency trading
IoT stream processing

Code Example (Java)

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

DataStream<Event> events = env
    .addSource(new FlinkKafkaConsumer<>("events", new EventSchema(), properties));

DataStream<Aggregation> aggregated = events
    .keyBy(Event::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(5)))
    .aggregate(new EventAggregator());

aggregated.addSink(new FlinkKafkaProducer<>("aggregated", new AggregationSchema(), properties));

env.execute("Real-Time Aggregation");

Configuration Recommendations

# flink-conf.yaml
taskmanager.memory.process.size: 4096m
taskmanager.numberOfTaskSlots: 4
state.backend: rocksdb
state.checkpoints.dir: s3://bucket/checkpoints
state.savepoints.dir: s3://bucket/savepoints
execution.checkpointing.interval: 60000

Apache Spark Streaming

Architecture

Micro-batch processing model
RDD-based (legacy) or DataFrame-based (Structured Streaming)
Integration with Spark batch ecosystem
Catalyst optimizer for SQL

Strengths

Unified batch and stream processing
Excellent Python support (PySpark)
Strong ML integration (MLlib, feature stores)
Interactive SQL analytics
Mature ecosystem (Delta Lake, Databricks)
Good for complex transformations

Weaknesses

Higher latency (seconds vs milliseconds)
Less suitable for real-time use cases
Larger resource footprint

Best Use Cases

ETL pipelines (batch + streaming)
Machine learning feature engineering
Data lake ingestion and transformation
Analytics with SQL (Structured Streaming)
Hybrid batch/stream workloads
Python-first data science teams

Code Example (Python)

from pyspark.sql import SparkSession
from pyspark.sql.functions import window, col

spark = SparkSession.builder \
    .appName("StreamingETL") \
    .getOrCreate()

# Read from Kafka
events = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "events") \
    .load()

# Parse JSON and aggregate
aggregated = events \
    .selectExpr("CAST(value AS STRING) as json") \
    .select(from_json(col("json"), schema).alias("data")) \
    .groupBy(window("data.timestamp", "5 minutes"), "data.user_id") \
    .count()

# Write to data lake
query = aggregated.writeStream \
    .format("delta") \
    .outputMode("append") \
    .option("checkpointLocation", "s3://bucket/checkpoints") \
    .start("s3://bucket/aggregated")

query.awaitTermination()

Configuration Recommendations

# Spark configuration
spark.conf.set("spark.sql.streaming.checkpointLocation", "s3://bucket/checkpoints")
spark.conf.set("spark.sql.shuffle.partitions", "100")
spark.conf.set("spark.streaming.kafka.maxRatePerPartition", "1000")
spark.conf.set("spark.sql.streaming.stateStore.providerClass",
               "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")

Kafka Streams

Architecture

Client library (not separate cluster)
Embedded in application JVM
Stateful processing with RocksDB
Exactly-once processing support

Strengths

Simple deployment (no cluster management)
Exactly-once semantics
Tight Kafka integration
Automatic load balancing (consumer group protocol)
Low operational overhead

Weaknesses

Kafka-only (cannot read from other sources)
Java/Scala only
Scales with application instances
Limited ecosystem compared to Flink/Spark

Best Use Cases

Microservices stream processing
Real-time aggregations
Stateful transformations
Event enrichment
Stream-stream joins
Applications already using Kafka

Code Example (Java)

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

StreamsBuilder builder = new StreamsBuilder();

KStream<String, Event> events = builder.stream("events");

KTable<Windowed<String>, Long> aggregated = events
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
    .count();

aggregated.toStream().to("aggregated");

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

Configuration Recommendations

# Kafka Streams config
processing.guarantee=exactly_once_v2
num.stream.threads=4
state.dir=/tmp/kafka-streams
cache.max.bytes.buffering=10485760
commit.interval.ms=1000

ksqlDB

Architecture

Server-based SQL engine
Built on Kafka Streams
Push and pull queries
Materialized views

Strengths

SQL interface (familiar to analysts)
No code required for simple transformations
Materialized views for serving queries
Integration with Confluent Platform
Good for rapid prototyping

Weaknesses

Limited to SQL expressiveness
Performance overhead vs native Kafka Streams
Smaller community than other options
Confluent-specific ecosystem

Best Use Cases

Real-time dashboards (SQL queries)
Simple transformations and aggregations
Analyst self-service analytics
Prototyping stream processing logic
Materialized view serving

Code Example (ksqlDB)

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

-- Aggregate events in 5-minute windows
CREATE TABLE aggregated AS
SELECT
  user_id,
  WINDOWSTART AS window_start,
  COUNT(*) AS event_count
FROM events
WINDOW TUMBLING (SIZE 5 MINUTES)
GROUP BY user_id
EMIT CHANGES;

-- Push query (continuous)
SELECT * FROM aggregated EMIT CHANGES;

-- Pull query (point-in-time)
SELECT * FROM aggregated WHERE user_id = 'user-123';

Comparison Matrix

Processing Model

Feature	Flink	Spark	Kafka Streams	ksqlDB
Processing Model	True streaming	Micro-batch	True streaming	True streaming
Latency	Millisecond	Second	Millisecond	Millisecond
Throughput	Very High	Very High	High	Medium
Event Time	Native	Supported	Native	Supported
Watermarks	Built-in	Manual	Built-in	Automatic

State Management

Feature	Flink	Spark	Kafka Streams	ksqlDB
State Backend	RocksDB, In-Memory	RocksDB, HDFS	RocksDB	Kafka Streams
Checkpointing	Excellent	Good	Good	Automatic
Fault Tolerance	Exactly-once	Exactly-once	Exactly-once	Exactly-once
State Size	Large (TB+)	Large (TB+)	Medium (100s GB)	Medium

Deployment

Feature	Flink	Spark	Kafka Streams	ksqlDB
Deployment	Cluster	Cluster	Embedded	Server
Scaling	Task slots	Executors	App instances	Server instances
Operations	Medium	Medium	Low	Low
Resource Usage	Medium	High	Low	Medium

Ecosystem

Feature	Flink	Spark	Kafka Streams	ksqlDB
Language Support	Java, Scala, PyFlink	Java, Scala, Python, R	Java, Scala	SQL
Connectors	Many (Flink CDC)	Many (Spark connectors)	Kafka-only	Kafka-only
ML Integration	FlinkML	MLlib (Excellent)	None	None
SQL Support	Table API, SQL	Structured Streaming	None	Native
Community	Large	Very Large	Large	Medium

Selection Flowchart

Primary Decision Path

START: What is primary requirement?

├─ Latency
│  ├─ Millisecond-level? → FLINK or KAFKA STREAMS
│  ├─ Sub-second? → FLINK
│  └─ Seconds OK? → SPARK
│
├─ Processing Model
│  ├─ True streaming? → FLINK or KAFKA STREAMS
│  ├─ Micro-batch OK? → SPARK
│  └─ SQL interface? → KSQLDB
│
├─ Deployment
│  ├─ Embedded in app? → KAFKA STREAMS
│  ├─ Separate cluster? → FLINK or SPARK
│  └─ Serverless? → KSQLDB
│
└─ Language
   ├─ Python required? → SPARK
   ├─ Java/Scala only? → FLINK or KAFKA STREAMS
   └─ SQL only? → KSQLDB

Use Case Alignment

START: What is use case?

├─ Real-Time Analytics
│  ├─ Millisecond dashboards? → FLINK
│  ├─ SQL interface for analysts? → KSQLDB
│  └─ Complex aggregations? → SPARK
│
├─ ETL Pipelines
│  ├─ Batch + streaming? → SPARK
│  ├─ Streaming only? → FLINK
│  └─ Simple transformations? → KAFKA STREAMS
│
├─ Machine Learning
│  ├─ Feature engineering? → SPARK (MLlib)
│  ├─ Model inference? → FLINK or SPARK
│  └─ Training? → SPARK
│
├─ Complex Event Processing
│  ├─ Pattern matching? → FLINK (CEP)
│  ├─ Stateful operations? → FLINK
│  └─ Simple filters? → KAFKA STREAMS
│
└─ Microservices
   ├─ Embedded processing? → KAFKA STREAMS
   ├─ Shared infrastructure? → FLINK
   └─ SQL-based? → KSQLDB

Operational Constraints

START: What are operational needs?

├─ Team Skills
│  ├─ Python expertise? → SPARK
│  ├─ Java/Scala expertise? → FLINK or KAFKA STREAMS
│  ├─ SQL-first team? → KSQLDB
│  └─ Mixed skills? → SPARK
│
├─ Infrastructure
│  ├─ Kubernetes? → FLINK or SPARK
│  ├─ Serverless? → KSQLDB or managed services
│  ├─ On-prem? → FLINK or SPARK
│  └─ Cloud-native? → Managed FLINK or SPARK
│
└─ Budget
   ├─ Cost-sensitive? → KAFKA STREAMS (no cluster)
   ├─ Enterprise support? → SPARK (Databricks)
   └─ Open source only? → FLINK or KAFKA STREAMS

Technology-Specific Guidance

When to Choose Flink

Strong indicators:

Real-time analytics with millisecond SLAs
Complex Event Processing (CEP) requirements
Event-time processing critical (out-of-order events)
Large state management (100s GB to TBs)
High-frequency event streams

Example scenarios:

Fraud detection (sub-second response)
Real-time recommendation systems
IoT sensor data processing
Network monitoring and alerting
Financial trading systems

Code smell (avoid Flink):

Python-first team with no Java/Scala expertise
Simple transformations (overkill)
Batch-heavy workloads

When to Choose Spark Streaming

Strong indicators:

Batch + streaming hybrid workloads
Machine learning integration (MLlib, feature stores)
Python-first data science teams
Data lake ingestion and transformation
Interactive SQL analytics

Example scenarios:

ETL pipelines (ingest to data warehouse)
ML feature engineering pipelines
Log aggregation and analysis
Batch reprocessing with streaming updates
Data quality monitoring

Code smell (avoid Spark):

Real-time requirements (<1 second latency)
Simple event routing (overkill)

When to Choose Kafka Streams

Strong indicators:

Microservices architecture
No separate cluster desired
Tight Kafka integration
Stateful transformations
Event enrichment

Example scenarios:

Order processing pipeline
User session aggregation
Event enrichment from database
Stream-stream joins
Real-time inventory updates

Code smell (avoid Kafka Streams):

Need to process from multiple sources (not just Kafka)
Python requirement
Large-scale batch processing

When to Choose ksqlDB

Strong indicators:

SQL-first team (analysts, not engineers)
Rapid prototyping
Simple aggregations and transformations
Materialized views for serving
No custom code desired

Example scenarios:

Real-time dashboard queries
Simple aggregation pipelines
Analyst self-service analytics
Prototyping before coding
Materialized view serving layer

Code smell (avoid ksqlDB):

Complex business logic (SQL limitations)
Performance-critical paths
Need for custom UDFs

Performance Tuning

Flink Tuning

# Parallelism
parallelism.default: 4
taskmanager.numberOfTaskSlots: 4

# Memory
taskmanager.memory.process.size: 4096m
taskmanager.memory.managed.fraction: 0.4

# Checkpointing
execution.checkpointing.interval: 60000
execution.checkpointing.mode: EXACTLY_ONCE
state.backend.incremental: true

# RocksDB state backend
state.backend.rocksdb.block.cache-size: 256m
state.backend.rocksdb.write-buffer-size: 64m

Spark Tuning

# Shuffle partitions
spark.conf.set("spark.sql.shuffle.partitions", "200")

# Memory
spark.conf.set("spark.executor.memory", "4g")
spark.conf.set("spark.executor.memoryOverhead", "1g")

# Streaming
spark.conf.set("spark.sql.streaming.stateStore.providerClass",
               "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider")
spark.conf.set("spark.streaming.kafka.maxRatePerPartition", "1000")

Kafka Streams Tuning

# Parallelism
num.stream.threads=4

# State store
state.dir=/mnt/kafka-streams
cache.max.bytes.buffering=10485760

# Processing
commit.interval.ms=1000
processing.guarantee=exactly_once_v2

# RocksDB
rocksdb.config.setter=CustomRocksDBConfig

Migration Strategies

From Spark to Flink

Reasons to migrate:

Need lower latency (millisecond vs second)
Event-time processing critical
True streaming model required

Migration approach: 1. Identify Spark Streaming jobs 2. Rewrite using Flink DataStream API 3. Run both systems in parallel (shadow mode) 4. Compare results and performance 5. Cutover to Flink

From Kafka Streams to Flink

Reasons to migrate:

Need to process from multiple sources (not just Kafka)
Large state exceeds Kafka Streams capacity
Complex CEP patterns required

Migration approach: 1. Extract business logic to shared library 2. Implement Flink job with same logic 3. Dual-write results for validation 4. Cutover consumer to Flink results 5. Decommission Kafka Streams app

From ksqlDB to Code

Reasons to migrate:

SQL limitations (complex business logic)
Performance requirements
Need custom UDFs

Migration approach: 1. Export ksqlDB queries as reference 2. Implement equivalent logic in Flink/Kafka Streams 3. Test with same input data 4. Compare outputs 5. Cutover

Conclusion

Default recommendation: Start with Kafka Streams for microservices, Flink for real-time analytics, Spark for batch+stream hybrid.

Real-time requirements: Choose Flink for millisecond latency and complex event processing.

Python teams: Choose Spark Streaming for excellent PySpark support and ML integration.

Operational simplicity: Choose Kafka Streams for embedded processing without cluster management.

SQL interface: Choose ksqlDB for analyst self-service and rapid prototyping.

Related skills

Lark BaseLet their coding agents read from, write to, and query structured data stored in Lark Base (Feishu) spreadsheets and databases.493k

Azure KustoRun fast KQL queries against Azure Data Explorer for log analytics, telemetry, and time-series data.484k1.3k

Lark SheetsCreate, read, update, and export spreadsheets inside Lark (Feishu) directly from their AI coding agent.389k15.9k

Supabase Postgres Best PracticesAutomatically apply Supabase Postgres performance rules during query writing, schema design, and database optimization.313k2.4k

Firebase Data ConnectGenerate complete, production-ready Firebase Data Connect GraphQL schemas and queries for complex relational data models.112k392

Firebase FirestoreAdd Cloud Firestore read/write capabilities to an Android Kotlin app using the official Firebase SDK.77k392

FAQ

What is the difference between a message broker and a stream processor?

Brokers like Kafka store and distribute event streams with durability and replay; stream processors like Flink transform, aggregate and join streaming data with windowing and stateful operations.

Which delivery guarantee should I use by default?

At-least-once is the default for most applications; it never loses messages but requires idempotent consumers.

Databasespipelinesetlanalytics

About

Streaming Data by the numbers

streaming-data capabilities & compatibility

What streaming-data says it does

Add your badge

What it does

Who is it for?

When should I use this skill?

What you get

By the numbers

Files

Streaming Data Processing

When to Use This Skill

Core Concepts

Message Brokers vs Stream Processors

Delivery Guarantees

Quick Start Guide

Step 1: Choose a Message Broker

Step 2: Choose a Stream Processor (if needed)

Step 3: Implement Producer/Consumer Patterns

Common Patterns

Basic Producer Pattern

Basic Consumer Pattern

Error Handling Strategy

Decision Frameworks

Framework: Message Broker Selection

Framework: Stream Processor Selection

Framework: Language Selection

Advanced Patterns

Event Sourcing

Change Data Capture (CDC)

Exactly-Once Processing

Error Handling

Reference Files

Decision Guides

Language-Specific Implementation

Advanced Topics

Validation Scripts

Validate Kafka Configuration

Generate Schema Registry Templates

Benchmark Throughput

Code Examples

TypeScript Example (KafkaJS)

Python Example (confluent-kafka-python)

Go Example (kafka-go)

Java Example (Apache Kafka)

Technology Comparison

Message Broker Comparison

Stream Processor Comparison

Client Library Recommendations

Related Skills

Troubleshooting

Consumer Lag Issues

Message Loss

Duplicate Messages

Performance Bottlenecks

Message Broker Selection Guide

Table of Contents

Overview

Apache Kafka

Architecture

Strengths

Weaknesses

Best Use Cases

Configuration Recommendations

Apache Pulsar

Architecture

Strengths

Weaknesses

Best Use Cases

Configuration Recommendations

Redpanda

Architecture

Strengths

Weaknesses

Best Use Cases

Configuration Recommendations

RabbitMQ

Architecture

Strengths