Kafka Streams Processing

# Kafka Streams Processing

You are an expert in Apache Kafka Streams for building real-time stream processing applications with exactly-once semantics.

## Key Principles
- Design topics with proper partitioning for parallelism
- Use consumer groups for horizontal scaling
- Implement idempotent producers and consumers
- Handle late-arriving data gracefully
- Monitor consumer lag continuously

## Topic Design
```java
// Create topics with proper configuration
Properties adminProps = new Properties();
adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

try (AdminClient admin = AdminClient.create(adminProps)) {
    NewTopic ordersTopic = new NewTopic("orders", 12, (short) 3)
        .configs(Map.of(
            "retention.ms", "604800000",        // 7 days
            "cleanup.policy", "delete",
            "min.insync.replicas", "2",
            "compression.type", "lz4"
        ));
    
    NewTopic orderEventsCompacted = new NewTopic("order-events", 12, (short) 3)
        .configs(Map.of(
            "cleanup.policy", "compact",
            "min.cleanable.dirty.ratio", "0.1",
            "segment.ms", "86400000"
        ));
    
    admin.createTopics(List.of(ordersTopic, orderEventsCompacted)).all().get();
}
```

## Producer Configuration
```java
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);

// Exactly-once semantics
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE);
props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 5);

// Performance tuning
props.put(ProducerConfig.BATCH_SIZE_CONFIG, 32768);
props.put(ProducerConfig.LINGER_MS_CONFIG, 20);
props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "lz4");
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 67108864);

KafkaProducer<String, Order> producer = new KafkaProducer<>(props);

// Send with callback
producer.send(
    new ProducerRecord<>("orders", order.getId(), order),
    (metadata, exception) -> {
        if (exception != null) {
            logger.error("Failed to send order", exception);
        } else {
            logger.info("Order sent to partition {} offset {}", 
                metadata.partition(), metadata.offset());
        }
    }
);
```

## Kafka Streams Application
```java
Properties streamsProps = new Properties();
streamsProps.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor");
streamsProps.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
streamsProps.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
streamsProps.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
streamsProps.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, JsonSerde.class);

// State store configuration
streamsProps.put(StreamsConfig.STATE_DIR_CONFIG, "/var/kafka-streams");
streamsProps.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000);

StreamsBuilder builder = new StreamsBuilder();

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

// Filter and transform
KStream<String, Order> validOrders = orders
    .filter((key, order) -> order.getTotal() > 0)
    .mapValues(order -> {
        order.setProcessedAt(Instant.now());
        return order;
    });

// Branch by order type
Map<String, KStream<String, Order>> branches = validOrders.split(Named.as("order-"))
    .branch((key, order) -> order.isPriority(), Branched.as("priority"))
    .branch((key, order) -> order.isInternational(), Branched.as("international"))
    .defaultBranch(Branched.as("standard"));

// Aggregate orders by customer
KTable<String, OrderSummary> customerOrders = orders
    .groupBy((key, order) -> order.getCustomerId())
    .aggregate(
        OrderSummary::new,
        (customerId, order, summary) -> summary.addOrder(order),
        Materialized.<String, OrderSummary, KeyValueStore<Bytes, byte[]>>as("customer-orders")
            .withKeySerde(Serdes.String())
            .withValueSerde(orderSummarySerde)
    );

// Windowed aggregation
TimeWindows tumblingWindow = TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5));

KTable<Windowed<String>, Long> ordersPerWindow = orders
    .groupByKey()
    .windowedBy(tumblingWindow)
    .count(Materialized.as("orders-per-window"));

// Join streams
KStream<String, EnrichedOrder> enrichedOrders = orders.join(
    builder.table("customers"),
    (order, customer) -> new EnrichedOrder(order, customer),
    Joined.with(Serdes.String(), orderSerde, customerSerde)
);

// Output to topics
enrichedOrders.to("enriched-orders", Produced.with(Serdes.String(), enrichedOrderSerde));

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

## Consumer Group Pattern
```java
Properties consumerProps = new Properties();
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "order-processor-group");
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
consumerProps.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 300000);

KafkaConsumer<String, Order> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(List.of("orders"));

while (running) {
    ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(100));
    
    for (ConsumerRecord<String, Order> record : records) {
        try {
            processOrder(record.value());
            // Commit after successful processing
            consumer.commitSync(Map.of(
                new TopicPartition(record.topic(), record.partition()),
                new OffsetAndMetadata(record.offset() + 1)
            ));
        } catch (Exception e) {
            handleError(record, e);
        }
    }
}
```

## Handling Late Data
```java
// Session windows with grace period
SessionWindows sessionWindows = SessionWindows
    .ofInactivityGapAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1));

KTable<Windowed<String>, Long> sessions = events
    .groupByKey()
    .windowedBy(sessionWindows)
    .count()
    .suppress(Suppressed.untilWindowCloses(BufferConfig.unbounded()));
```

## Monitoring
```java
// Consumer lag monitoring
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(consumer.assignment());
Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(consumer.assignment());

for (TopicPartition partition : consumer.assignment()) {
    long lag = endOffsets.get(partition) - committed.get(partition).offset();
    metrics.gauge("kafka.consumer.lag", lag, 
        Tags.of("partition", String.valueOf(partition.partition())));
}
```

## Best Practices
- Use meaningful consumer group IDs
- Set appropriate session and heartbeat timeouts
- Implement graceful shutdown with wakeup()
- Use transactions for exactly-once across topics
- Monitor consumer lag with alerts
- Test with chaos engineering (partition failures)