Real-Time Stream Processing

Real-time stream processing handles data flows that must be processed within milliseconds or seconds of arrival, rather than waiting for batch windows. This approach is essential when downstream systems require immediate responses to events, such as fraud detection, real-time recommendations, or live analytics dashboards.

Processing Events as They Arrive

To consume a stream of events, configure your processor to connect to a message broker or streaming platform. The typical pattern involves creating a consumer that reads from a topic or partition and applies transformations.

Reading From a Stream

# Example using standard Kafka CLI tools
kafka-console-consumer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic user-events \
  --from-beginning \
  --value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

The --from-beginning flag ensures you read all available messages, not just new ones. Without it, consumers may miss events that arrived while they were offline.

Managing Offsets

Each consumer group tracks its progress through a topic via offsets. Mismanagement leads to duplicate processing or data loss.

# View current offset positions
kafka-run-class.sh kafka.tools.GetOffsetShell \
  --broker-list kafka.example.com:9092 \
  --topic user-events \
  --time -1

The --time -1 flag fetches the latest offset for each partition. Use --time -2 to get the earliest available offset.

Processing Multiple Topics

For systems that aggregate from multiple sources:

# Example consumer configuration
kafka-console-consumer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic user-events,user-actions \
  --group event-aggregator \
  --value-deserializer=org.apache.kafka.common.serialization.StringDeserializer \
  --property print.key=true

Use the --group flag to define a consumer group. Multiple consumers in the same group will split partitions among themselves.

Handling Data Transformations

Transformations occur within the processing pipeline, not as separate batch jobs. A stream processor must handle each event individually, maintaining state where necessary.

Basic Transformation Logic

# Example Python-based stream processor
import json

def process_event(event_data):
    try:
        event = json.loads(event_data)
        # Apply transformations
        transformed = {
            'user_id': event['user_id'],
            'timestamp': event['timestamp'],
            'processed_at': time.time(),
            'event_type': event['type']
        }
        return json.dumps(transformed)
    except Exception as e:
        # Log failed events for later replay
        print(f"Failed to process: {e}")
        return None

Events that fail processing should be routed to a dead-letter queue, not dropped silently.

Stateful Processing

Stateful transformations require maintaining data between events:

// Example in Java using Kafka Streams
StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("input-topic");

KTable<String, Long> counts = source
  .mapValues(value -> value.toLowerCase())
  .groupBy(KeyValue.pair("user", "action"))
  .count();

counts.toStream().to("output-topic");

Stateful operations use RocksDB internally and require sufficient disk space. The --state.dir parameter controls where state is stored.

Error Handling and Recovery

Stream processors must handle failures gracefully, ensuring no data is lost during processing interruptions.

Dead Letter Queues

Failed messages should be sent to a separate topic:

# Producer configuration for failed events
kafka-console-producer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic failed-events \
  --property parse.key=true \
  --property key.separator=:

The --property parse.key=true flag enables key-based routing. When a transformation fails, the original event should be preserved in the dead-letter queue.

Idempotent Processing

Ensure that processing the same event twice does not produce duplicate results:

# Example idempotent consumer
def process_with_idempotency(event_key, event_data):
    # Check if we've already processed this event
    if redis.exists(f"processed:{event_key}"):
        return
    
    # Process the event
    result = transform_event(event_data)
    
    # Mark as processed
    redis.setex(f"processed:{event_key}", 3600, "1")
    return result

Use a time-based cache or database to track which events have been processed.

Managing Backpressure

When processing speed lags behind arrival rate, backpressure must be handled gracefully.

Throttling Configuration

# Consumer with rate limiting
kafka-console-consumer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic high-volume-topic \
  --max.poll.records=100 \
  --poll.interval.ms=1000

The --max.poll.records parameter limits how many messages are fetched per poll cycle. This prevents overwhelming downstream systems.

Buffer Management

# Producer with buffer settings
kafka-console-producer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic buffered-topic \
  --buffer.memory=33554432 \
  --retries=3

The --buffer.memory parameter sets how much memory to use for buffering. If this is too small, producers may block waiting for space.

Monitoring and Alerting

Stream processors must expose metrics for operational visibility:

Health Checks

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

The output shows Lag which indicates how many messages behind the consumer is. If this value grows continuously, it signals a processing problem.

Metrics Endpoint

Many stream processors expose metrics via HTTP:

# Example metrics endpoint
curl http://processor.example.com/metrics

Look for metrics like throughput, error rates, and latency. The response format depends on the implementation but typically returns JSON or Prometheus-style text.

Failure Modes and Pitfalls

Silent Data Loss

Stream processors may silently drop events when they encounter configuration errors:

# Wrong deserializer causes silent failures
kafka-console-consumer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic user-events \
  --value-deserializer=org.apache.kafka.common.serialization.IntegerDeserializer

If events are JSON strings but the deserializer expects integers, messages will be skipped without error reporting.

Offset Commit Issues

The --enable.auto.commit flag should generally be disabled for production:

# Producer with manual offset control
kafka-console-consumer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic user-events \
  --enable.auto.commit=false \
  --isolation.level=read_committed

When auto-commit is enabled, a consumer might commit an offset before successfully processing all messages in that batch.

Resource Exhaustion

Stream processors can exhaust memory or CPU:

# Monitor consumer resource usage
top -p $(pgrep -f kafka-console-consumer)

Check for high CPU usage or memory leaks in long-running consumers. Large stateful operations may require tuning JVM parameters.

Scaling and Deployment

Horizontal Scaling

Add more consumers to a consumer group to increase throughput:

# Start multiple consumers
kafka-console-consumer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic user-events \
  --group scaling-group \
  --threads 4

The --threads parameter increases parallelism within a single consumer process.

Deployment Considerations

Stream processors should be designed for rolling updates:

# Example deployment script
kubectl set image deployment/stream-processor processor=stream-processor:v2.1

Use readiness probes to ensure new instances are fully initialized before accepting traffic. The --wait flag in Kubernetes deployments ensures proper startup sequencing.

Partition Management

Ensure your topics have sufficient partitions for parallel processing:

# Check topic configuration
kafka-topics.sh \
  --bootstrap-server kafka.example.com:9092 \
  --describe \
  --topic user-events

Look for the PartitionCount field. Too few partitions limit horizontal scaling capability.

Configuration Best Practices

Producer Settings

# Producer with optimal settings
kafka-console-producer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic reliable-topic \
  --acks=1 \
  --retries=3 \
  --batch.size=16384

The --acks=1 setting ensures messages are written to at least one replica before acknowledging. Use --retries=3 for transient failures.

Consumer Settings

# Consumer with safe settings
kafka-console-consumer.sh \
  --bootstrap-server kafka.example.com:9092 \
  --topic critical-topic \
  --group safety-group \
  --enable.auto.commit=false \
  --max.poll.interval.ms=300000 \
  --session.timeout.ms=10000

The --max.poll.interval.ms parameter controls how long consumers can take between poll operations before being considered dead. Set this to at least 5 minutes for long-running processing.

Stream processing requires careful attention to data flow, error handling, and system boundaries. The key is maintaining consistency while allowing for real-time responsiveness.

This page was rewritten on 16 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.