Kafka Streams

You are a Kafka Streams architect. The user wants to build a stream processing application using the Kafka Streams DSL to process real-time event data with stateless and stateful operations.

What to check first

• Verify Kafka broker is running on localhost:9092 (or your configured bootstrap server) with kafka-broker-api-versions.sh --bootstrap-server localhost:9092
• Confirm Kafka topics exist or will be auto-created: check kafka-topics.sh --list --bootstrap-server localhost:9092
• Ensure Java 11+ and Maven/Gradle are installed with java -version and mvn -version

Steps

1. Add Kafka Streams dependency to pom.xml: version should match your Kafka broker version (e.g., 3.6.0)
2. Create a StreamsBuilder instance — this is the DSL entry point for topology construction
3. Define the source topology using builder.stream() or builder.table() for your input topic(s)
4. Apply stateless transformations: map(), filter(), flatMap(), or branch() based on your processing logic
5. For stateful operations, use aggregate(), reduce(), or count() with an explicit state store name
6. Configure windowing if needed: TimeWindows.of(), SessionWindows.with(), or SlidingWindows.withTimeDifference()
7. Write results to output topic(s) using .to() or .toStream().to() depending on whether you're publishing a KStream or KTable
8. Build the topology with builder.build(), configure StreamsConfig properties, and create/start a KafkaStreams instance
9. Add shutdown hooks to gracefully close the streams application

Code

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.TimeWindows;

import java.util.Properties;
import java.time.Duration;

public class OrderProcessingApp {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "order-processor");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes

Note: this example was truncated in the source. See the GitHub repo for the latest full version.

Common Pitfalls

• Treating this skill as a one-shot solution — most workflows need iteration and verification
• Skipping the verification steps — you don't know it worked until you measure
• Applying this skill without understanding the underlying problem — read the related docs first

When NOT to Use This Skill

• When a simpler manual approach would take less than 10 minutes
• On critical production systems without testing in staging first
• When you don't have permission or authorization to make these changes

How to Verify It Worked

• Run the verification steps documented above
• Compare the output against your expected baseline
• Check logs for any warnings or errors — silent failures are the worst kind

Production Considerations

• Test in staging before deploying to production
• Have a rollback plan — every change should be reversible
• Monitor the affected systems for at least 24 hours after the change