Kafka Streams Basics

@Configuration
@EnableKafkaStreams
class StreamConfig {
    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    KafkaStreamsConfiguration streamsConfig() {
        return new KafkaStreamsConfiguration(Map.of(
            StreamsConfig.APPLICATION_ID_CONFIG,    "order-processor",
            StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker:9092",
            StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG,   Serdes.String().getClass(),
            StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass()
        ));
    }
}

@Component
class OrderTopology {
    @Autowired
    void buildPipeline(StreamsBuilder builder) {
        KStream<String, String> orders = builder.stream("orders-raw");

        orders
            .filter((key, value) -> value.contains(""status":"PAID""))
            .mapValues(value -> enrichOrder(value))
            .to("orders-enriched");   // sink topic
    }
}

@Autowired
void buildPipeline(StreamsBuilder builder) {
    KStream<String, Order> orders = builder.stream(
        "orders",
        Consumed.with(Serdes.String(), orderSerde)
    );

    // Count orders per customer — stateful aggregation
    KTable<String, Long> orderCounts = orders
        .groupByKey()
        .count(Materialized.as("order-counts-store"));  // named state store

    orderCounts.toStream().to("order-counts-topic",
        Produced.with(Serdes.String(), Serdes.Long()));

    // KStream-KTable join: enrich order with customer data
    KTable<String, Customer> customers = builder.table("customers");
    orders.join(customers,
        (order, customer) -> enrich(order, customer))
        .to("orders-enriched");
}

# Kafka Streams fault tolerance and scaling:

# 1. State store backed by changelog topic (created automatically)
#    Internal topic: <app-id>-<state-store-name>-changelog
#    e.g.: order-processor-order-counts-store-changelog

# 2. Restart recovery:
#    App starts → consumer group rebalance → partitions assigned
#    → state store restored from changelog (or standby replica if configured)
#    → processing resumes from last committed offset

# 3. Standby replicas — warm spare state (reduces restore time)
num.standby.replicas=1  # keep 1 standby copy in another instance

# 4. Scaling: run 2 instances of the same application-id
#    → Kafka Streams splits partitions between them
#    → each instance handles its own slice of state

# 5. Interactive Queries — read state store directly (no external DB)
ReadOnlyKeyValueStore<String, Long> store =
    streams.store(StoreQueryParameters.fromNameAndType(
        "order-counts-store", QueryableStoreTypes.keyValueStore()));
Long count = store.get("customer-42");