Apache Kafka Stream Processing: 5 Core Patterns for Building Million-TPS Real-Time Data Pipelines

Apache Kafka Stream Processing: The Cornerstone of Real-Time Data Pipelines

Microservice data silos, high batch processing latency, difficult real-time analytics — core challenges facing enterprise data infrastructure. Apache Kafka as a distributed event streaming platform, with million-TPS throughput, millisecond latency, and Exactly-Once semantics, has become the de facto standard for real-time data pipelines. In 2026, Kafka stream processing is widely deployed in financial trading, IoT telemetry, and user behavior analytics.

This article covers 5 core patterns, guiding you through Producer/Consumer → Kafka Streams → Connect → Schema Registry → production tuning.

---

Core Concepts

Concept	Description
Kafka	Distributed event streaming platform
Topic	Message category, logical partition collection
Partition	Topic's physical shard, parallelism unit
Consumer Group	Consumer group for load balancing
Kafka Streams	Kafka native stream processing library
Kafka Connect	Data connector framework
Schema Registry	Avro/Protobuf Schema management service
Exactly-Once	Exactly-once semantics avoiding duplicate processing

---

Problem Analysis: 5 Major Kafka Stream Processing Challenges

1. Partition strategy selection: Wrong partitioning causes data skew
2. Consumer Rebalance: Consumer group rebalancing causes consumption pauses
3. Exactly-Once implementation: End-to-end exactly-once semantics configuration is complex
4. Schema evolution: Avro Schema change compatibility management
5. Monitoring and alerting: Lag monitoring and consumer delay warnings

---

Step-by-Step: 5 Kafka Stream Processing Patterns

Pattern 1: High-Throughput Producer and Consumer

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
props.put(ProducerConfig.BATCH_SIZE_CONFIG, "32768");
props.put(ProducerConfig.LINGER_MS_CONFIG, "10");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
for (int i = 0; i < 1_000_000; i++) {
    producer.send(new ProducerRecord<>("orders",
        String.valueOf(i % 64), "{\"orderId\":" + i + "}"));
}
producer.flush();

Pattern 2: Kafka Streams Topology and Windowed Aggregation

StreamsBuilder builder = new StreamsBuilder();

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

KTable<Windowed<String>, Long> orderCounts = orders
    .groupBy((key, value) -> extractUserId(value))
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
    .count(Materialized.as("order-counts"));

orderCounts.toStream()
    .map((windowedKey, count) -> new KeyValue<>(
        windowedKey.key(),
        "{\"userId\":\"" + windowedKey.key() + "\",\"count\":" + count + "}"))
    .to("order-counts-output");

Properties streamsProps = new Properties();
streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-analytics");
streamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
streamsProps.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");

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

Pattern 3: Kafka Connect Data Pipeline

{
  "name": "postgres-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "dbz",
    "database.dbname": "orders_db",
    "database.server.name": "pgserver1",
    "table.include.list": "public.orders,public.customers"
  }
}

Pattern 4: Schema Registry and Avro Serialization

Properties avroProps = new Properties();
avroProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
avroProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
avroProps.put("schema.registry.url", "http://schema-registry:8081");

Schema orderSchema = SchemaBuilder.record("Order")
    .namespace("com.example")
    .fields()
    .requiredLong("orderId")
    .requiredString("userId")
    .requiredDouble("amount")
    .endRecord();

GenericRecord order = new GenericData.Record(orderSchema);
order.put("orderId", 12345L);
order.put("userId", "user-001");
order.put("amount", 99.99);

Pattern 5: Production Monitoring and Tuning

# Consumer Lag monitoring
kafka-consumer-groups --bootstrap-server kafka:9092 \
  --describe --group order-processor

# Key metrics
# kafka.producer.record-send-rate
# kafka.producer.record-error-rate
# kafka.consumer.consumer-lag
# kafka.server.UnderReplicatedPartitions

---

Pitfall Guide

Pitfall 1: Partition key causing data skew

// ❌ Wrong: timestamp as partition key, all data in one partition
producer.send(new ProducerRecord<>("orders", String.valueOf(System.currentTimeMillis()), data));

// ✅ Correct: business key hash partitioning
producer.send(new ProducerRecord<>("orders", orderId, data));

Pitfall 2: Consumer auto-commit offset

// ❌ Wrong: auto-commit, messages may be lost
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");

// ✅ Correct: manual commit after processing
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
consumer.commitSync();

Pitfall 3: Not handling Rebalance

// ❌ Wrong: no rebalance listener
consumer.subscribe(List.of("orders"));

// ✅ Correct: register rebalance listener
consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override
    public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
        consumer.commitSync();
    }
});

Pitfall 4: Incompatible Schema changes

// ❌ Wrong: removing required field breaks backward compatibility
// ✅ Correct: add new fields with defaults, maintain compatibility
Schema newSchema = SchemaBuilder.record("Order")
    .fields()
    .requiredLong("orderId")
    .optionalString("newField")
    .endRecord();

Pitfall 5: Not configuring Exactly-Once

// ❌ Wrong: default At-Least-Once, may process duplicates
// ✅ Correct: enable Exactly-Once v2
streamsProps.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, "exactly_once_v2");

---

Error Troubleshooting

#	Error	Cause	Solution
1	NotLeaderOrFollowerException	Partition leader switch	Producer auto-retry, check Broker health
2	CommitFailedException	Rebalance caused offset commit failure	Reduce max.poll.interval.ms or increase processing speed
3	SerializationException	Schema mismatch	Check Schema Registry compatibility
4	TimeoutException	Request timeout	Increase request.timeout.ms and delivery.timeout.ms
5	RecordTooLargeException	Message exceeds max.message.bytes	Increase Broker's message.max.bytes
6	GroupAuthorizationException	Consumer group insufficient permissions	Check ACL configuration
7	TopicExistsException	Topic already exists	Use --if-not-exists or check Topic config
8	WakeupException	Consumer woken up for shutdown	Normal shutdown, call consumer.wakeup()
9	RebalanceInProgressException	Rebalance in progress	Wait for rebalance to complete
10	SchemaRegistryException	Schema Registry unreachable	Check schema.registry.url and network

---

Advanced Optimization

1. Tiered Storage: Cold data auto-migration to S3, reducing Broker storage costs
2. KRaft Mode: Remove ZooKeeper dependency, simplify operations
3. Idempotent Producer + Transactions: End-to-end Exactly-Once semantics
4. Consumer Lag Auto-Scaling: Dynamically adjust Consumer instances based on Lag
5. MirrorMaker 2 Cross-Cluster Replication: Multi-region disaster recovery and migration

---

Comparison

Dimension	Kafka	Pulsar	RabbitMQ	Redis Streams
Throughput	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐
Latency	⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐
Exactly-Once	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐	⭐⭐
Stream Processing	⭐⭐⭐⭐⭐	⭐⭐⭐⭐	⭐⭐	⭐⭐
Ecosystem	⭐⭐⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐	⭐⭐⭐
Ops Complexity	⭐⭐⭐	⭐⭐⭐	⭐⭐⭐⭐⭐	⭐⭐⭐⭐⭐

---

Summary: Apache Kafka stream processing, with million-TPS throughput and Exactly-Once semantics, is the de facto standard for real-time data pipelines. Kafka suits scenarios requiring high-throughput real-time data processing, especially financial trading, IoT telemetry, and user behavior analytics. With KRaft mode maturing and tiered storage普及 in 2026, Kafka's operational costs continue to decrease.

---

Online Tools

• JSON Formatter: /en/json/format
• Hash Calculator: /en/encode/hash
• cURL to Code: /en/dev/curl-to-code