Data Streaming

Data Streaming

Purpose

Design streaming data pipelines with Kafka topic architecture, Flink/Kafka Streams processing, schema evolution, and reliability guarantees.

Agent Protocol

Trigger

Exact user phrases: "streaming", "Kafka", "Flink", "Kinesis", "stream processing", "event stream", "real-time", "CDC", "change data capture", "message queue", "stream architecture", "Kafka topic", "Kafka consumer", "stream processing pipeline", "exactly-once", "schema registry", "Avro", "Debezium", "ksqlDB".

Input Context

Before activating, verify:

• Streaming platform (Kafka, Kinesis, Pulsar)
• Processing framework (Flink, Kafka Streams, ksqlDB, Spark Streaming)
• Source systems (database CDC, application events, IoT, logs)
• Target systems (warehouse, lake, search index, cache)
• Throughput and latency requirements
• Data volume and retention requirements
• Security and compliance needs
• Team expertise with streaming technologies

Output Artifact

Streaming pipeline design with topic model, processing logic, reliability config as YAML and SQL.

Response Format

# Topic topology
# Partition strategy
# Consumer group config

-- ksqlDB query

// Flink job skeleton

No preamble. No postamble. No explanations. No filler/hedging/transitions. Compress output — why use many token when few do trick.

Completion Criteria

• Topic model defined with partition count and retention
• Schema Registry configured with evolution rules
• Stream processing job with exactly-once semantics
• CDC pipeline from source database configured
• Error handling with DLQ defined
• Monitoring and lag alerting configured
• Security configured (TLS, auth, ACLs)
• Data retention and compaction strategy defined

Max Response Length

300 lines of code and configuration.

Kafka Architecture

Brokers and Clusters

A Kafka cluster consists of multiple brokers. Each broker is a server that stores topic partitions and serves produce/consume requests. Brokers are identified by a unique ID. A cluster typically has 3-7 brokers for production. Controller broker handles partition leadership and cluster metadata. Brokers should have sufficient disk (RAID 10, SSDs) and network bandwidth for the expected throughput.

Topics and Partitions

A topic is a logical stream of messages. Each topic has multiple partitions for parallelism. Messages within a partition are ordered and assigned an offset. Partitions are distributed across brokers for fault tolerance. Partition count determines the maximum parallelism for consumers. Keyed messages go to the same partition (hash(key) % partition_count). Partitions are the unit of parallelism and replication.

Replication and ISR

Each partition has a leader (handles all reads/writes) and followers (replicate from leader). In-Sync Replicas (ISR) are followers that are fully caught up with the leader. The min.insync.replicas config sets the minimum number of replicas that must acknowledge a write. acks=all means the leader waits for all in-sync replicas to acknowledge. If a follower falls behind (replication lag > replica.lag.time.max.ms), it is removed from the ISR.

Consumer Group Rebalancing

A consumer group divides topic partitions among its members. When a consumer joins, leaves, or fails, the group rebalances. Eager rebalancing stops all consumers and reassigns all partitions (causes a pause). Cooperative rebalancing reassigns incrementally (minimal pause). Static group membership (group.instance.id) provides stable assignment across restarts. Monitor rebalance frequency — >10 rebalances per hour indicates instability.

Exactly-Once Semantics in Kafka

Producer Semantics

acks=0: fire-and-forget, no acknowledgment, possible data loss. acks=1: leader acknowledges, possible leader failover loss. acks=all (with min.insync.replicas): leader + ISR acknowledge, no loss. Enable enable.idempotence=true to prevent duplicate produces within a session. Set transactional.id for exactly-once across partitions.

Consumer Semantics

At-most-once: commit offset before processing (message may be lost on failure). At-least-once: commit offset after processing (message may be reprocessed on failure). Exactly-once: process and commit offset atomically via transactional API. Use isolation.level=read_committed to only read committed messages. Combine with idempotent sinks (upsert, idempotent operations) for pragmatic exactly-once.

Transactional API

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(record1);
    producer.send(record2);
    producer.sendOffsetsToTransaction(offsets, consumerGroup);
    producer.commitTransaction();
} catch (Exception e) {
    producer.abortTransaction();
}

Kafka Streams / ksqlDB

Kafka Streams

Kafka Streams is a lightweight library for stream processing within a Java application. It processes one record at a time. State stores provide fault-tolerant key-value state. Streams topology defines the processing DAG. Exactly-once semantics via processing.guarantee=exactly_once_v2. Kafka Streams is best for simple transformations, filtering, enrichment, and KTable joins within a single application.

Kafka Streams Topology Components

Source processor: reads from input topics. Stream processor: transforms, filters, maps data. Stateful processor: maintains state (aggregations, joins, windowing). Sink processor: writes to output topics. GlobalKTable broadcast tables: for enrichment with reference data that fits in memory. KTable changelog topics: for materialized state backed by compacted topics.

StreamsBuilder builder = new StreamsBuilder();
KStream<String, Order> orders = builder.stream("orders", Consumed.with(Serdes.String(), orderSerde));

KTable<String, Customer> customers = builder.table("customers", Consumed.with(Serdes.String(), customerSerde));

orders.join(customers, (order, customer) -> {
    order.setCustomerName(customer.getName());
    return order;
}).filter((key, order) -> order.getAmount() > 100)
  .to("enriched-orders", Produced.with(Serdes.String(), orderSerde));

ksqlDB

ksqlDB provides a SQL interface to Kafka Streams. Create STREAM and TABLE definitions on Kafka topics. Use SQL to filter, join, aggregate, and materialize streams. Pull queries return a single result (like traditional SQL). Push queries continuously stream results. ksqlDB is best for teams that prefer SQL, rapid prototyping, and simple streaming ETL.

-- Define streams and tables
CREATE STREAM orders (
    order_id VARCHAR KEY,
    customer_id VARCHAR,
    amount DOUBLE,
    product VARCHAR
) WITH (KAFKA_TOPIC='orders', VALUE_FORMAT='AVRO');

CREATE TABLE customers (
    customer_id VARCHAR PRIMARY KEY,
    name VARCHAR,
    email VARCHAR
) WITH (KAFKA_TOPIC='customers', VALUE_FORMAT='AVRO', KEY='customer_id');

-- Enrich and aggregate
CREATE STREAM enriched_orders AS
    SELECT o.order_id, c.name, o.amount, o.product
    FROM orders o LEFT JOIN customers c ON o.customer_id = c.customer_id;

CREATE TABLE daily_sales AS
    SELECT product, SUM(amount) AS total_sales, COUNT(*) AS order_count
    FROM orders WINDOW TUMBLING (SIZE 1 DAY)
    GROUP BY product
    EMIT CHANGES;

Flink Streaming

Event Time and Watermarks

Flink processes events based on event time (the time the event occurred) rather than processing time. Watermarks track the progress of event time — a watermark with timestamp T means no more events with event time < T will arrive. Watermarks handle out-of-order events. Set the watermark based on the maximum observed event latency. Lateness beyond the watermark is handled via allowed lateness and side outputs.

Watermark Strategies

Periodic watermarks: emitted at configurable intervals based on observed event timestamps (BoundedOutOfOrdernessTimestampExtractor). Punctuated watermarks: emitted per event when certain conditions are met. Ideal watermark strategy: watermark = max_observed_timestamp - expected_lateness. Set expected_lateness based on source-system SLAs and observed behavior.

DataStream<Order> orders = env
    .addSource(kafkaConsumer)
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((event, timestamp) -> event.getEventTime())
    );

Windowing

Tumbling windows: fixed-size, non-overlapping — window(TumblingProcessingTimeWindows.of(Time.hours(1))). Hopping windows: fixed-size, overlapping — window(SlidingProcessingTimeWindows.of(Time.minutes(10), Time.minutes(5))) (window size 10 min, slide 5 min). Session windows: group events by activity gap — window(ProcessingTimeSessionWindows.withGap(Time.minutes(30))). Sliding windows: gap-less continuous windows for trending. Each window type has specific use cases and performance characteristics.

Checkpointing

Flink checkpoints save the job state for failure recovery. Checkpoints are triggered periodically (every 60s recommended). State is stored in a durable backend (RocksDB for large state, Heap for low-latency). Checkpoints enable exactly-once semantics by restoring to the last successful checkpoint on failure. Savepoints are manually triggered checkpoints used for job upgrades, scaling, and maintenance.

# Flink config for exactly-once with checkpointing
state.backend: rocksdb
state.checkpoints.dir: s3://flink-checkpoints/job-name
state.checkpoints.num-retained: 5
execution.checkpointing.interval: 60s
execution.checkpointing.min-pause: 30s
execution.checkpointing.tolerable-failed-checkpoints: 3
execution.checkpointing.timeout: 10min
execution.checkpointing.mode: EXACTLY_ONCE

# Unaligned checkpoints for high-throughput jobs
execution.checkpointing.unaligned: true
execution.checkpointing.unaligned.max-buffers: 100

Partitioning Strategy

Choose partition count based on desired throughput: partitions = desired_throughput / single_partition_throughput. Key-based partitioning ensures order per key. Round-robin distributes load evenly. Partition count affects parallelism — each partition is consumed by at most one consumer in a group. Too few partitions limits parallelism; too many increases overhead. Rebalancing partitions later is operationally expensive — plan upfront.

Schema Registry

Avro / Protobuf Schemas

Schema Registry stores and validates schemas for Kafka topics. Producers register schemas; consumers fetch schemas to deserialize. Schema ID is included in the message header for efficient wire format. Compatibility modes: BACKWARD (new schema reads old data — default), FORWARD (old schema reads new data), FULL (both directions), NONE (no checks).

{
  "type": "record",
  "name": "OrderCreated",
  "namespace": "com.example.events",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "total_amount", "type": "double"},
    {"name": "currency", "type": "string", "default": "USD"},
    {"name": "items", "type": {"type": "array", "items": {
      "type": "record",
      "name": "OrderItem",
      "fields": [
        {"name": "product_id", "type": "string"},
        {"name": "quantity", "type": "int"},
        {"name": "unit_price", "type": "double"}
      ]
    }}}
  ]
}

Schema Evolution Rules

Never remove fields. New fields must have defaults for backward compatibility. Use union types for optional fields: ["null", "string"]. Document schema changes in the schema metadata. Evolve schemas in the registry before updating producers/consumers. Test schema changes against existing data before deploying. Use logical types for precise semantics (decimal, date, timestamp-millis).

Streaming Sources and Sinks

Source Connectors

Sink Connectors

CDC with Debezium

Debezium Architecture

Debezium connects to database transaction logs (WAL for PostgreSQL, binlog for MySQL, redo log for Oracle, change feed for SQL Server). Emits each row change as a separate Kafka message. Handles schema changes, snapshots, and continuous streaming.

Debezium Message Structure

{
  "op": "c",         // c=create, u=update, d=delete, r=snapshot
  "before": null,
  "after": {
    "id": 123,
    "name": "Alice",
    "email": "alice@example.com"
  },
  "source": {
    "db": "postgres",
    "table": "customers",
    "lsn": 12345678,
    "ts_ms": 1717200000000
  },
  "ts_ms": 1717200000100
}

Debezium Configuration

{
  "name": "postgres-orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres-prod",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${POSTGRES_PASSWORD}",
    "database.dbname": "orders_db",
    "database.server.name": "postgres",
    "table.include.list": "public.orders,public.customers",
    "plugin.name": "pgoutput",
    "publication.name": "debezium_pub",
    "publication.autocreate.mode": "filtered",
    "slot.name": "debezium_slot",
    "tombstones.on.delete": "false",
    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "schema.registry.url": "http://schema-registry:8081",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState"
  }
}

Error Handling Patterns

Dead Letter Queue (DLQ)

Every processing step routes failures to a DLQ. DLQ topics named {source-topic}.dlq. DLQ messages include original payload, error message, stack trace, and processing timestamp. DLQ monitored for growth and reviewed weekly. Backfill script reprocesses DLQ records after schema fix.

topics:
  dlq: "orders.created.v1.dlq"
  dlq_config:
    partitions: 3
    replication: 3
    retention: 90d
    cleanup.policy: delete

Deserialization Error Handling

// Flink: side output for deserialization errors
OutputTag<String> deserializationErrors = new OutputTag<String>("deserialization-errors") {};

DataStream<Order> validOrders = kafkaSource
    .process(new DeserializeWithErrorHandling(deserializationErrors));

validOrders.getSideOutput(deserializationErrors)
    .addSink(new KafkaSink<String>(dlqTopic));

Retry Strategy

Retryable: connection timeout, rate limit, lock wait, temporary error. Non-retryable: schema mismatch, data validation, permission denied. Retry strategy: exponential backoff (1s, 2s, 4s, 8s, 16s) up to 5 attempts. Circuit breaker after 5 consecutive failures on same partition. Use Flink's built-in restart strategy: fixed delay or exponential delay.

Ordering Guarantees

Partition-Level Ordering

Messages within a partition are strictly ordered by offset. Key-based partitioning ensures related messages (same key) go to the same partition — guaranteeing order for that key. Without key, round-robin distribution provides no ordering guarantee across partitions. Consumer processes partitions in offset order.

Global Ordering

Single partition topic = total ordering (but limits parallelism to 1). Partition key on all messages with same constant value = all messages in one partition (defeats parallelism). Practical approach: accept per-partition ordering, design consumers to handle out-of-order events by using event timestamps.

Handling Out-of-Order Events in Processing

Buffer events in state store with configured grace period. Use event time processing with watermarks instead of processing time. Deduplicate by event ID with sliding window state. Handle late events via side outputs for offline reconciliation.

State Management

State Store Types

State TTL Configuration

# Flink state TTL configuration
table.exec.state.ttl: 1h  # SQL API
# DataStream API
stateTtlConfig = StateTtlConfig
    .newBuilder(Time.hours(24))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .build();

State Migration

State schema changes (e.g., adding a field to state) require savepoint-to-savepoint migration. Options: (1) stop job, apply migration on state, restart; (2) write migration logic in ProcessFunction; (3) start new job with fresh state and dual-read source. Prefer savepoint-based upgrade with schema evolution where possible.

Testing Strategies

Unit Testing

Test individual operators and functions with mock Kafka records. Use TopologyTestDriver for Kafka Streams, TestHarness for Flink processors. Verify: correct outputs for given inputs, side outputs for errors, state updates, watermark behavior.

Integration Testing

Spin up Kafka + Schema Registry + processing engine via Testcontainers or docker-compose. Produce test events, consume results, assert correctness. Test: consumer group rebalancing, checkpointing and recovery (kill job, verify resume), schema evolution compatibility, exactly-once delivery.

Performance Testing

Define throughput targets (msg/s, MB/s), latency SLAs (p50, p95, p99), and data loss tolerance. Test with production-scale data volume and partition count. Identify bottlenecks: CPU-bound (serialization, compression), memory-bound (state size, buffering), IO-bound (network, disk). Run for minimum 24 hours to detect long-term drift.

Security

Encryption

In-transit: TLS between all components (brokers, clients, ZooKeeper, Schema Registry). In-rest: disk encryption on broker storage. End-to-end: application-level encryption for sensitive fields using envelope encryption.

Authentication

SSL client authentication: mutual TLS between clients and brokers. SASL/PLAIN: username/password (simple, needs TLS). SASL/SCRAM: salt-challenge authentication (recommended). SASL/GSSAPI (Kerberos): enterprise integration. SASL/OAUTHBEARER: OAuth2 token-based.

Authorization

Kafka ACLs: --allow-principal User:app1 --operation read --topic orders. Topic-level: read, write, create, describe, alter. Consumer group-level: read, describe. Cluster-level: create topics, describe configs. Use prefix ACLs for topic patterns: --topic orders.*. Prefer RBAC via Apache Ranger for multi-team clusters.

Monitoring

Consumer Lag

Consumer lag is the difference between the latest produced offset and the consumer's committed offset. High lag means the consumer is falling behind. Lag is the most critical streaming metric. Monitor lag every 60 seconds. Alert on lag > 1000 messages or lag growing steadily (indicates consumer cannot keep up).

Key Metrics

Producer metrics: request rate, error rate, compression ratio, batch size. Consumer metrics: lag, poll rate, processing time, commit rate. Broker metrics: request rate, disk usage, network throughput, ISR count, under-replicated partitions. Flink metrics: checkpoint duration, state size, records processed per second, latency. All metrics should feed into a monitoring dashboard with alerts for anomalous values.

Streaming Health Dashboard

Orders Streaming Pipeline
  ┌──────────────────────────────────────────┐
  │ Status: ✅ Healthy                        │
  │ End-to-end latency: 245ms                │
  │ Consumer lag: 23 (max 100)              │
  │ Throughput: 1,420 msg/s                  │
  │ Checkpoint: 12s (last successful: 12s ago)│
  │ Error rate: 0.02%                        │
  │ State size: 4.2 GB                       │
  └──────────────────────────────────────────┘

Streaming vs Batch Decision Guide

Common Streaming Topology Patterns

CDC Pipeline Pattern

Source DB → Debezium → Kafka Topic (raw CDC events) → Flink Transform → 
  Kafka Topic (cleaned events) → ksqlDB Materialized View → Sink (warehouse, cache, search)

Event Sourcing Pattern

Producer → Kafka Topic (domain events) → Multiple Consumer Groups:
  ├── Group A: Flink → State Store → Materialized View
  ├── Group B: Kafka Streams → KTable → Join with other streams
  └── Group C: ksqlDB → Push Query → Real-time Dashboard

Microservices Communication Pattern

Service A → Kafka Topic (command/event) → Service B
  ├── Service B processes event → Publishes result event
  └── Service A subscribes to result event → Reacts accordingly

Log Aggregation Pattern

Application Instances → Kafka Topic (logs) → Flink → 
  ├── Elasticsearch (indexed logs)
  ├── S3 (archived logs)
  └── Alert System (error threshold detection)

Fan-Out Pattern

Input Topic → Kafka Streams / Flink
  ├── Transform A → Topic A → Consumer Group A
  ├── Transform B → Topic B → Consumer Group B
  └── Transform C → Topic C → Consumer Group C

Kafka Topic Naming Convention Details

Retention Policy Decision Guide

Partition Count Calculation Examples

Streaming SLA Targets

Streaming Anti-Patterns

Topic Explosion

Creating a topic per event type per entity (hundreds of topics). Leads to ZK overhead, management chaos, partition imbalance. Fix: group related events into logical topics with type discriminator field.

Missing Schema Validation

Producing/consuming raw JSON without Schema Registry. Inevitably leads to deserialization failures when schema changes. Fix: enforce Schema Registry with Avro/Protobuf and BACKWARD compatibility.

Ignoring Offset Management

Auto-committing offsets with enable.auto.commit=true means processing state may not match committed offset. Fix: manual offset commits after processing complete, or use transactional API.

Synchronous Processing in Consumers

Calling external APIs synchronously in consumer poll loop blocks the thread and causes rebalance timeouts. Fix: async processing with callbacks, or use a separate processing thread pool.

Static Partition Count

Setting partition count once and never reviewing it. As throughput grows, partitions become a bottleneck. Fix: design for growth (start with 3x current needs) or use tiered storage (Pulsar) that abstracts partitions.

Streaming Platform Ecosystem

Apache Pulsar

Pulsar is a multi-tenant, high-throughput messaging platform with native geo-replication. Unlike Kafka, Pulsar separates serving (Brokers) from storage (Bookies via Apache BookKeeper), enabling elastic scaling without data rebalancing. Key features: segment-centric storage for unlimited log retention, tiered storage (offload to S3/GCS), built-in Pulsar Functions for lightweight processing, and native multi-tenancy. Topics are virtual and scale transparently. Use Pulsar for geo-distributed deployments, unlimited retention, or multi-tenant streaming.

Redpanda

Redpanda is a Kafka-compatible streaming platform in C++ with a single binary (no ZooKeeper, no JVM). Achieves 10x lower latency and 6x higher throughput per node. Uses Raft-based consensus for HA, full Kafka API compatibility, and includes built-in Schema Registry, REST Proxy, and Connectors. Best for teams wanting Kafka compatibility with reduced ops overhead and lower TCO.

Streaming Databases (Materialize, RisingWave)

Both provide incremental materialized views on streaming data using PostgreSQL-compatible SQL. Materialize updates results incrementally as new data arrives without re-running queries, with persistent storage and exactly-once semantics. RisingWave is cloud-native with decoupled compute-storage and object store persistence. Both support CREATE MATERIALIZED VIEW on streams, window functions, stream-table joins, and JDBC/PostgreSQL wire protocol. Choose Materialize for Kafka-native SQL with strong consistency; RisingWave for large-scale persistence with PG compatibility.

Rules

• Exactly-once semantics for all critical streams
• Schema Registry with Avro, backward compatibility
• Partition count based on throughput, not convenience
• Consumer group lag monitored every 60 seconds
• DLQ for every processing step
• No production schema change without compatibility check
• Compacted topics for keyed state, delete retention for events
• Watermarks account for out-of-order events
• Alert on lag > 1000 or lag growing for 5+ minutes
• Set retention based on replay and audit requirements
• Never auto-commit offsets in production
• Test checkpointing by simulating broker failures
• Monitor rebalance frequency as cluster health indicator
• Use TLS for all inter-component communication
• Every topic must have documented owner and retention policy

References

• references/flink-streaming.md — Flink Streaming
• references/kafka-architecture.md — Kafka Architecture
• references/pulsar-patterns.md — Apache Pulsar Patterns
• references/streaming-architecture.md — Streaming Architecture
• references/streaming-databases.md — Streaming Databases
• references/streaming-monitoring.md — Streaming Monitoring

Handoff

data-data-warehouse for streaming data landing in the warehouse data-etl-pipeline for batch processing of streamed data