MMIT_INTERVAL_MS_CONFIG, 1000);

    StreamsBuilder builder = new StreamsBuilder();
    KStream<String, String> rawEvents = builder.stream("events-raw");
    
    // Parse, filter, and aggregate in a 5-minute tumbling window
    KTable<Windowed<String>, Long> sessionCounts = rawEvents
        .filter((k, v) -> v != null && v.contains("session_id"))
        .groupBy((key, value) -> value.split(",")[0]) // group by session_id
        .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
        .count(Materialized.as("session-count-store"));
        
    sessionCounts.toStream().to("sessions-aggregated");
    
    KafkaStreams streams = new KafkaStreams(builder.build(), props);
    streams.start();
    Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
}

}

### Step 3: State Management & Windowing
- **State Backend**: Kafka Streams uses RocksDB by default. Tune `cache.max.bytes.buffering`, `rocksdb.config.class`, and compaction strategies for write-heavy workloads.
- **Windowing**: Tumbling windows for fixed intervals, sliding windows for overlapping aggregations, session windows for gap-based activity.
- **Late Data**: Use `.grace()` or `.suppress()` to handle out-of-order events without violating exactly-once guarantees.

### Step 4: Fault Tolerance & Checkpointing
- Exactly-once v2 relies on Kafka transactions. Producers use `transactional.id`, consumers read with `isolation.level=read_committed`.
- State stores are changelog-backed to internal Kafka topics. Recovery time equals state size / partition throughput.
- Enable `state.dir` on fast local storage. Never place state on network mounts.

### Step 5: Architecture Decisions
| Decision | Recommendation | Rationale |
|----------|----------------|-----------|
| Pull vs Push | Pull (Kafka consumer model) | Predictable backpressure, consumer-controlled rate |
| Stateful vs Stateless | Stateful only when required | Stateless scales linearly; state introduces rebalancing cost |
| Partitioning Strategy | Key-based with cardinality analysis | Prevents hot partitions; aim for 2-4x partition count vs expected peak throughput |
| Schema Evolution | Protobuf with forward compatibility | Reduces payload size; avoids deserialization failures during rolling deploys |
| Backpressure Handling | Consumer lag monitoring + auto-scaling | Kafka Streams scales with partition count, not thread count |

## Pitfall Guide

1. **Ignoring Watermarks & Late Data**
   Processing events without grace periods or suppression causes incomplete window results and state corruption. Always define late-data handling explicitly.

2. **Misconfiguring Exactly-Once**
   Enabling `EXACTLY_ONCE_V2` without idempotent downstream sinks or transactional producers creates silent data duplication. Verify sink compatibility before deployment.

3. **Treating Streams as Unbounded Batches**
   Applying batch thinking (e.g., `collect()`, `toLocalIterator()`) on infinite streams triggers OOM errors. Use bounded windows, incremental aggregations, and state stores.

4. **Partition Skew & Hot Keys**
   Uneven key distribution causes single-partition bottlenecks. Profile key cardinality, apply salting for high-frequency keys, and monitor consumer lag per partition.

5. **State Store Misconfiguration**
   Default RocksDB settings assume general workloads. Tune `write_buffer_size`, `max_background_compactions`, and `level_compaction_dynamic_level_bytes` for write-heavy streaming.

6. **Schema Drift Without Compatibility Checks**
   Adding required fields or changing types breaks deserialization across rolling deployments. Enforce schema registry policies and use forward/backward compatibility modes.

7. **Inadequate Backpressure & Lag Monitoring**
   Streaming systems fail silently when consumer lag grows. Monitor `records-lag-max`, `commit-rate`, and `state-maintenance-time`. Alert on lag > threshold for >2 minutes.

## Production Bundle

### Action Checklist
- [ ] Enforce schema registry with compatibility checks on all input/output topics
- [ ] Configure exactly-once v2 with transactional producers and read_committed consumers
- [ ] Implement grace periods or suppression for late-arriving events
- [ ] Tune RocksDB state backend for write amplification and compaction
- [ ] Partition topology aligned with key cardinality and expected peak throughput (2-4x ratio)
- [ ] Deploy continuous lag monitoring, checkpoint success rate, and state migration metrics
- [ ] Test partition rebalancing under load; verify state store restoration time < SLA
- [ ] Run chaos tests: broker failure, network partition, schema rollback, and consumer group reset

### Decision Matrix
| Framework | Latency (p99) | State Complexity | Ops Overhead | Ecosystem Maturity | Learning Curve | Best Use Case |
|-----------|---------------|------------------|--------------|--------------------|----------------|---------------|
| Kafka Streams | 10-50ms | Medium (embedded) | Low (library) | High | Low | Embedded processing, microservices |
| Apache Flink | 5-30ms | High (managed) | High (cluster) | High | Medium-High | Complex state, CEP, high throughput |
| Spark Structured Streaming | 500ms-2s | Medium | Medium | High | Medium | Batch/stream unification, ML pipelines |
| Materialize | 1-10ms | Low (SQL-driven) | Low | Medium | Low | Real-time materialized views, analytics |

### Configuration Template
**docker-compose.yml**
```yaml
version: '3.8'
services:
  kafka:
    image: confluentinc/cp-kafka:7.5.0
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
    ports: ["9092:9092"]
  schema-registry:
    image: confluentinc/cp-schema-registry:7.5.0
    environment:
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: PLAINTEXT://kafka:9092
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
    ports: ["8081:8081"]

application.properties

application.id=realtime-pipeline-v1
bootstrap.servers=localhost:9092
default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde
processing.guarantee=exactly_once_v2
commit.interval.ms=500
consumer.isolation.level=read_committed
producer.enable.idempotence=true
state.dir=/var/lib/kafka-streams
rocksdb.config.class=org.apache.kafka.streams.state.RocksDBConfigSetter

Quick Start Guide

1. Spin Up Infrastructure: Run docker-compose up -d. Verify Kafka and Schema Registry are healthy via kafka-topics --list and /subjects endpoint.
2. Generate Test Data: Publish 10k events to events-raw using kafka-console-producer or a custom generator. Ensure keys map to session IDs and payloads follow the expected schema.
3. Deploy Processor: Build and run the Kafka Streams application. Monitor logs for StreamsConfig validation, state store initialization, and partition assignment.
4. Validate & Observe: Consume from sessions-aggregated to verify windowed counts. Track consumer lag via kafka-consumer-groups.sh --describe. Confirm exactly-once by injecting duplicates and verifying idempotent output.

Real-time processing is not a tool choice; it is an architectural contract. Latency, state, and fault tolerance must be designed explicitly. Teams that align partitioning, schema evolution, and exactly-once semantics from day one avoid the most common production failures. The pipeline outlined above provides a deterministic foundation. Scale it by adding partition-aware routing, state backend tuning, and continuous observability.