Real-time data processing

log (Kafka/Pulsar). 2. Replayability: Processing state can be rebuilt by replaying the log from any offset. 3. Stateful Processing: Stream processors maintain local state for windowing and aggregation, backed by distributed storage.

Step-by-Step Implementation

1. Ingestion and Schema Enforcement

Events must be ingested with strict schema validation to prevent pipeline corruption. Use a Schema Registry with Avro or Protobuf.

2. Stream Processing Engine

Implement a stateful processor. For TypeScript environments, libraries like node-rdkafka combined with in-memory state or embedded databases like RocksDB (via native bindings) are effective.

3. Windowing Strategy

Select windowing based on business logic:

• Tumbling Windows: Fixed size, non-overlapping (e.g., 1-minute aggregations).
• Sliding Windows: Fixed size, overlapping (e.g., 5-minute window sliding every 1 minute).
• Session Windows: Dynamic size based on activity gaps (e.g., user session tracking).

4. Idempotent Sinks

Downstream systems must handle duplicate events. Use idempotency keys derived from event metadata.

TypeScript Implementation: Stateful Windowed Aggregation

The following example demonstrates a stateful stream processor in TypeScript using a conceptual stream library. It implements a tumbling window with state management and watermarking for out-of-order event handling.

import { StreamProcessor, Window, Watermark, StateStore } from 'stream-processor-lib';

interface UserEvent {
    userId: string;
    timestamp: number; // Event time in ms
    action: 'click' | 'purchase';
    value: number;
}

interface AggregationState {
    count: number;
    totalValue: number;
    lastUpdated: number;
}

// Configuration
const WINDOW_SIZE_MS = 60_000; // 1 minute
const WATERMARK_DELAY_MS = 5_000; // Allow 5s late events

class RealTimeAnalyticsProcessor extends StreamProcessor<UserEvent> {
    private stateStore: StateStore<string, AggregationState>;
    private currentWatermark: Watermark;

    constructor() {
        super();
        this.stateStore = new StateStore<string, AggregationState>('user-aggregations');
        this.currentWatermark = new Watermark(Date.now() - WATERMARK_DELAY_MS);
    }

    async processEvent(event: UserEvent): Promise<void> {
        // 1. Update Watermark based on event time
        if (event.timestamp > this.currentWatermark.value + WATERMARK_DELAY_MS) {
            this.currentWatermark.advance(event.timestamp - WATERMARK_DELAY_MS);
            await this.flushWindows(this.currentWatermark);
        }

        // 2. Check if event is within valid window
        const windowStart = this.getWindowStart(event.timestamp);
        if (event.timestamp < this.currentWatermark.value) {
            // Late event handling: drop or route to dead letter
            console.warn(`Late event dropped: ${event.userId} @ ${event.timestamp}`);
            return;
        }

        // 3. Update State
        const stateKey = `${event.userId}:${windowStart}`;
        const currentState = await this.stateStore.get(stateKey) || {
            count: 0,
            totalValue: 0,
            lastUpdated: event.timestamp
        };

        currentState.count += 1;
        currentState.totalValue += event.value;
        currentState.lastUpdated = event.timestamp;

        await this.stateStore.put(stateKey, currentState);
    }

    private getWindowStart(timestamp: number): number {
        return Math.floor(timestamp / WINDOW_SIZE_MS) * WINDOW_SIZE_MS;
    }

    private async flushWindows(watermark: Watermark): Promise<void> {
        // Emit results for windows closed by the watermark
        const closedWindows = await this.stateStore.getKeysBefore(watermark.value);
        
        for (const key of closedWindows) {
            const state = await this.stateStore.get(key);
            if (state) {
                await this.emitResult({
                    windowKey: key,
                    count: state.count,
                    totalValue: state.totalValue,
                    processedAt: Date.now()
                });
                // Clean up state to prevent bloat
                await this.stateStore.delete(key);
            }
        }
    }

    private async emitResult(result: any): Promise<void> {
        // Publish to sink topic
        await this.publish('analytics-output', result);
    }
}

// Usage
const processor = new RealTimeAnalyticsProcessor();
processor.start();

Architecture Rationale:

• Watermarking: Handles out-of-order events by delaying window closure until the watermark advances. This ensures accuracy without infinite waiting.
• State Cleanup: State is deleted after window emission to prevent unbounded growth, a critical requirement for long-running processes.
• Type Safety: Interfaces enforce schema contracts, reducing runtime errors.

Pitfall Guide

1. Ignoring Backpressure

Mistake: Processing events faster than the sink can consume, leading to memory exhaustion or broker disconnections. Best Practice: Implement reactive backpressure. Throttle the consumer when sink latency spikes. Use circuit breakers to pause ingestion and buffer events in the broker rather than the application memory.

2. Partition Skew

Mistake: Choosing a partition key that concentrates traffic on specific brokers (e.g., partitioning by country when 80% of users are in one country). Best Practice: Analyze key distribution. Use composite keys or salting to distribute load. Monitor partition sizes and rebalance partitions dynamically if skew is detected.

3. State Bloat and TTL Misconfiguration

Mistake: Retaining state indefinitely or setting Time-To-Live (TTL) values too high, causing storage exhaustion. Best Practice: Define strict TTL policies aligned with business windows. Implement automated state compaction and monitor state store size metrics. Use RocksDB for efficient disk-based state management.

4. Clock Skew and Time Zones

Mistake: Relying on processing time instead of event time, or assuming synchronized clocks across distributed nodes. Best Practice: Always use event time for windowing. Implement watermarking to tolerate clock skew. Ensure all producers embed accurate timestamps. Use NTP synchronization for infrastructure clocks.

5. Exactly-Once Semantics Misconception

Mistake: Assuming "exactly-once" is free or trivial. It requires coordination between source, processor, and sink. Best Practice: Understand the cost. Exactly-once requires transactional sinks and idempotent writes. If the sink does not support transactions, implement idempotency keys in the payload. Document the consistency guarantees clearly.

6. Schema Evolution Breakage

Mistake: Changing event schemas without backward compatibility, breaking downstream consumers. Best Practice: Enforce schema compatibility rules in the registry (e.g., backward/forward compatibility). Version schemas. Implement schema migration strategies for in-flight events.

7. Dead Letter Queue Neglect

Mistake: Failing to handle poison pills (malformed events) that crash the processor repeatedly. Best Practice: Route deserialization errors and processing exceptions to a Dead Letter Queue (DLQ). Implement retry logic with exponential backoff for transient errors. Monitor DLQ depth and alert on spikes.

Production Bundle

Action Checklist

• Define Latency SLAs: Establish p50, p95, and p99 latency targets for each processing stage.
• Implement Schema Registry: Enforce Avro/Protobuf schemas with compatibility checks on all topics.
• Configure Idempotency: Ensure all sinks support idempotent writes using unique event IDs or composite keys.
• Deploy Dead Letter Queues: Create DLQ topics for all consumer groups with alerting on message accumulation.
• Monitor Consumer Lag: Set up alerts for consumer lag exceeding defined thresholds (e.g., >10k messages or >30s delay).
• Test Partition Skew: Load test with realistic data distributions to verify even partition utilization.
• Validate Watermarking: Simulate out-of-order events and late arrivals during integration tests.
• Disaster Recovery Drill: Test log replay capabilities to rebuild state from scratch within RTO limits.

Decision Matrix

Scenario	Recommended Approach	Why	Cost Impact
High-throughput IoT telemetry (>1M msg/s)	Kafka + Flink/Kafka Streams	Distributed state management and horizontal scaling handle massive throughput efficiently.	Medium-High (Compute intensive)
Low-latency fraud detection (<50ms)	In-memory Stream Processor (e.g., Redis Streams + custom logic)	Eliminates disk I/O latency; state is kept in memory for instant access.	Medium (Memory costs)
Complex event correlation (Pattern matching)	Stream Processing Engine with CEP	Native support for complex event processing patterns and windowed joins.	High (Complexity and compute)
Simple routing/Enrichment	Stateless Function (e.g., AWS Lambda, Cloudflare Workers)	No state management overhead; auto-scaling reduces cost for variable loads.	Low (Pay-per-invocation)
Regulatory compliance (Audit trail)	Kappa Architecture with immutable log	Full replayability and auditability of all events; single source of truth.	Medium (Storage costs)

Configuration Template

Kafka Producer Configuration (TypeScript / node-rdkafka)

import { Producer } from 'node-rdkafka';

export const createKafkaProducer = () => {
    const producer = new Producer({
        'client.id': 'realtime-processor',
        'bootstrap.servers': process.env.KAFKA_BROKERS || 'localhost:9092',
        'compression.codec': 'snappy', // Reduces bandwidth and storage
        'linger.ms': 10, // Batch small messages for throughput
        'batch.size': 1048576, // 1MB batch size
        'queue.buffering.max.messages': 10000,
        'message.send.max.retries': 3,
        'retry.backoff.ms': 100,
        'request.required.acks': 'all', // Ensures durability
        'enable.idempotence': true, // Prevents duplicates
        'transactional.id': 'processor-txn-01', // For exactly-once sinks
        'debug': 'broker,topic,msg', // Enable for troubleshooting
    });

    producer.connect();

    producer.on('event.error', (err) => {
        console.error('Kafka Producer Error:', err);
        // Implement circuit breaker or alerting
    });

    return producer;
};

Kubernetes Deployment Snippet for Stream Processor

apiVersion: apps/v1
kind: Deployment
metadata:
  name: stream-processor
spec:
  replicas: 3
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    spec:
      containers:
      - name: processor
        image: registry/stream-processor:latest
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        env:
        - name: KAFKA_BROKERS
          value: "kafka-broker-1:9092,kafka-broker-2:9092"
        - name: CONSUMER_GROUP_ID
          valueFrom:
            configMapKeyRef:
              name: processor-config
              key: group-id
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 5

Quick Start Guide

1. Initialize Local Environment: Run a local Kafka cluster using Docker Compose with KRaft mode (no Zookeeper).

   docker run -d --name kafka -p 9092:9092 -e KAFKA_NODE_ID=1 -e KAFKA_PROCESS_ROLES=broker,controller -e KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092 apache/kafka:latest
   

2. Create Topics: Create input and output topics with appropriate replication and partition counts.

   kafka-topics.sh --create --topic input-events --partitions 6 --replication-factor 1 --bootstrap-server localhost:9092
   kafka-topics.sh --create --topic output-analytics --partitions 6 --replication-factor 1 --bootstrap-server localhost:9092
   

3. Deploy Producer and Consumer: Use the TypeScript code provided in the Core Solution. Configure the producer to send events to input-events and the consumer to process and write to output-analytics. Ensure schema validation is active.

4. Verify Pipeline: Produce test events and monitor consumer lag using kafka-consumer-groups.sh. Validate that output events match expected aggregations and that latency remains within SLA targets. Check logs for watermark advancement and state cleanup.

Real-time data processing requires disciplined engineering. By adopting the Kappa architecture, implementing rigorous state management, and adhering to production best practices, backend systems can deliver the low-latency, high-reliability performance modern applications demand.