data-streamingUse this skill when asked about streaming, Kafka, Flink, Kinesis, stream processing, event stream, real-time, CDC, change data capture, message queue, or stream architecture. This skill enforces: Kafka topic design with partitioning strategy, Flink/ksqlDB stream processing with exactly-once semantics, Schema Registry with Avro/Protobuf, CDC integration with…
Independent GitHub snapshot · Automated classification · Original documentation governs
Design streaming data pipelines with Kafka topic architecture, Flink/Kafka Streams processing, schema evolution, and reliability guarantees.
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".
Before activating, verify:
Streaming pipeline design with topic model, processing logic, reliability config as YAML and SQL.
# 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.
300 lines of code and configuration.
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.
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.
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.
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.
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.
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.
producer.initTransactions();
try {
producer.beginTransaction();
producer.send(record1);
producer.send(record2);
producer.sendOffsetsToTransaction(offsets, consumerGroup);
producer.commitTransaction();
} catch (Exception e) {
producer.abortTransaction();
}
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.
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 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 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.
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())
);
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.
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
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 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"}
]
}}}
]
}
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).
| Source Type | Tool | Configuration Key |
|---|---|---|
| Database CDC | Debezium | connector.class=io.debezium.connector.postgresql.PostgresConnector |
| Application events | Kafka Producer | Direct API |
| Log files | Filebeat/Logstash → Kafka | Filebeat Kafka output |
| IoT/MQTT | MQTT Proxy → Kafka | Custom connector or bridge |
| HTTP webhooks | Kafka Connect HTTP source | HTTP source connector |
| Change tracking | JDBC Source Connector | mode=incrementing or timestamp |
| Sink Type | Tool | Configuration Key |
|---|---|---|
| Data warehouse | JDBC Sink, Snowflake Kafka Connector | Auto-create tables |
| Object storage | S3 Sink Connector | format.class=io.confluent.connect.s3.format.parquet.ParquetFormat |
| Search index | Elasticsearch Sink Connector | transforms=ExtractTopic |
| Cache | Redis Sink | Custom processor |
| Lakehouse | Delta/Iceberg Sink | upsert=true for CDC |
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.
{
"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
}
{
"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"
}
}
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
// 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));
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.
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.
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.
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.
| Store | Description | Best For |
|---|---|---|
| HashMapStateBackend | In-memory, fast, limited heap | Small state (< 1 GB), low-latency |
| RocksDBStateBackend | Disk-backed, large state | Large state (> 1 GB), moderate latency |
| FileSystemStateBackend | Checkpoints to FS | Development, testing |
# 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 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.
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.
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.
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.
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.
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.
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.
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).
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.
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 │
└──────────────────────────────────────────┘
| Requirement | Use Streaming | Use Batch |
|---|---|---|
| Latency | Sub-second to minutes | Hours to days |
| Data volume | Continuous, unbounded | Fixed, bounded |
| Processing model | Event-at-a-time or micro-batch | Full dataset |
| State management | Required (windowed, keyed) | Not needed |
| Failure recovery | Checkpoint/savepoint | Re-run from start |
| Cost | Higher (always-on infra) | Lower (scheduled compute) |
| Use case | Real-time dashboards, alerts, CDC | Reports, ML training, backfill |
Source DB → Debezium → Kafka Topic (raw CDC events) → Flink Transform →
Kafka Topic (cleaned events) → ksqlDB Materialized View → Sink (warehouse, cache, search)
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
Service A → Kafka Topic (command/event) → Service B
├── Service B processes event → Publishes result event
└── Service A subscribes to result event → Reacts accordingly
Application Instances → Kafka Topic (logs) → Flink →
├── Elasticsearch (indexed logs)
├── S3 (archived logs)
└── Alert System (error threshold detection)
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
| Convention | Example | Description |
|---|---|---|
| Domain events | orders.created.v1 | Business event |
| CDC events | postgres.orders.orders | Raw CDC from Debezium |
| Commands | cmd.ship-order.v1 | Command request |
| DLQ | orders.created.v1.dlq | Failed messages |
| Compacted | customer.profile.v1 | Latest state per key |
| Internal | _orders-aggregation-state | Application state store |
| Topic Type | Cleanup Policy | Retention | Example |
|---|---|---|---|
| Business events | delete | 7 days | orders.created.v1 |
| CDC events | delete | 30 days | postgres.orders.orders |
| Audit events | delete | 365 days | audit.access.v1 |
| Keyed state | compact | N/A (keep latest per key) | customer.profile.v1 |
| DLQ | delete | 90 days | orders.created.v1.dlq |
| Logs | delete | 3 days | app.logs.v1 |
| Scenario | Throughput/Sec | Msg Size | Recommended Partitions | Rationale |
|---|---|---|---|---|
| Order events | 500 msg/s | 2 KB | 6 | 500 * 2KB = 1MB/s |
| Clickstream | 50000 msg/s | 500 B | 24 | 50000 * 500B = 25MB/s |
| IoT sensor data | 100000 msg/s | 100 B | 48 | 100000 * 100B = 10MB/s |
| CDC from Postgres | 100 msg/s | 5 KB | 3 | Low volume |
| Audit log | 2000 msg/s | 1 KB | 12 | Moderate volume |
| Metric | Target | Alert Threshold | Measurement |
|---|---|---|---|
| End-to-end latency | < 1 second | > 5 seconds | Event creation to sink arrival |
| Consumer lag | < 100 | > 1000 | Difference between LEO and current offset |
| Checkpoint duration | < 30 seconds | > 60 seconds | Flink checkpoint metrics |
| Throughput per partition | < 5 MB/s | > 10 MB/s | JMX broker metrics |
| Error rate | < 0.1% | > 1% | Application metrics |
| Uptime | 99.99% | < 99.9% | Cluster and job health |
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.
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.
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.
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.
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.
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 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.
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.
data-data-warehouse for streaming data landing in the warehouse
data-etl-pipeline for batch processing of streamed data