Properties props = new Properties();
// APPLICATION_ID_CONFIG serves as the consumer group ID and the prefix
// for internal changelog topics. Choose a stable, descriptive name --
// changing it later means losing all committed offsets and state stores.
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-stream-app");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
// Serdes (Serializer/Deserializer) tell Kafka Streams how to convert
// bytes on the wire into Java objects. Mismatched serdes are the #1
// cause of cryptic deserialization errors in production.
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> source = builder.stream("input-topic");

// Filter keeps only records matching the predicate, then mapValues
// transforms the value without changing the key (preserving partitioning).
// Using mapValues instead of map is intentional -- map can re-partition
// data, which triggers an internal repartition topic and hurts performance.
source.filter((key, value) -> value.contains("important"))
      .mapValues(value -> value.toUpperCase())
      .to("output-topic");

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

// Word Count Example -- the "Hello World" of stream processing.
// Despite its simplicity, this demonstrates the full stateful pipeline:
// split -> rekey -> group -> aggregate -> materialize.
KStream<String, String> textLines = builder.stream("text-input");

KTable<String, Long> wordCounts = textLines
    // flatMapValues emits zero or more output records per input record.
    // Splitting text into words means one input line produces many output records.
    .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
    // groupBy re-keys the stream by the word itself. WARNING: this triggers
    // a repartition (internal topic + network shuffle). For high-volume streams,
    // this is the most expensive step in the pipeline.
    .groupBy((key, word) -> word)
    // count() creates a state store named "counts-store" backed by RocksDB.
    // The Materialized parameter makes this store queryable via Interactive Queries.
    .count(Materialized.as("counts-store"));

// Convert the KTable back to a KStream for writing to an output topic.
// KTable changes are emitted as update events (same key = overwrite).
wordCounts.toStream().to("word-counts-output", Produced.with(Serdes.String(), Serdes.Long()));

-- Create a stream over an existing Kafka topic.
-- This is a schema declaration, not a data copy. No data moves.
-- The stream is a continuous, unbounded view of events as they arrive.
CREATE STREAM user_clicks (
    user_id VARCHAR,     -- Maps to JSON field "user_id"
    url VARCHAR,         -- Maps to JSON field "url"
    timestamp VARCHAR    -- Maps to JSON field "timestamp"
) WITH (
    KAFKA_TOPIC = 'user_clicks_topic',   -- Must match an existing topic name
    VALUE_FORMAT = 'JSON'                -- Also supports AVRO, PROTOBUF
);

-- A TABLE in ksqlDB is the stream/table duality in action.
-- The underlying topic is treated as a changelog: for each user_id,
-- only the latest record matters. This is how you model "current state."
CREATE TABLE user_locations (
    user_id VARCHAR PRIMARY KEY,  -- PRIMARY KEY is required for tables
    city VARCHAR
) WITH (
    KAFKA_TOPIC = 'user_locations_topic',
    VALUE_FORMAT = 'JSON'
);

-- Enrich clicks with user location in real time.
-- This is one of the most powerful patterns in stream processing:
-- join a fast-moving event stream against a slowly-changing lookup table.
-- The result is a new stream (CREATE STREAM, not TABLE) because each
-- click event produces exactly one enriched output event.
CREATE STREAM enriched_clicks AS
SELECT 
    c.user_id, 
    c.url, 
    l.city
FROM user_clicks c
LEFT JOIN user_locations l ON c.user_id = l.user_id;

-- Count clicks per user per minute using a tumbling window.
-- Tumbling windows are fixed-size, non-overlapping time intervals.
-- A HOPPING window would overlap (e.g., 5-minute window every 1 minute).
-- A SESSION window groups events by inactivity gaps.
CREATE TABLE clicks_per_minute AS
SELECT 
    user_id, 
    COUNT(*) AS click_count
FROM user_clicks
WINDOW TUMBLING (SIZE 1 MINUTE)
GROUP BY user_id
EMIT CHANGES;   -- Emit updates as they happen, not just final results