How do you define a Kafka Streams topology (a KStream/KTable) as a Spring bean, and who starts it?

**The core idea.** With `@EnableKafkaStreams`, Spring gives you a single shared `StreamsBuilder` (from the default `StreamsBuilderFactoryBean`). Any `@Bean` method that declares a `StreamsBuilder` parameter receives that same instance; whatever nodes you add to it become part of one `Topology`. **KStream vs KTable (minimal definitions, since theory is a sibling topic — here just enough to wire correctly).** A `KStream<K,V>` is an unbounded stream where every record is an independent fact/event. A `KTable<K,V>` is a materialized changelog: for each key it holds the latest value (later records upsert earlier ones). You obtain them from the builder: `builder.stream("topic")` → `KStream`; `builder.table("topic")` → `KTable`. Spring's role stops at supplying the builder and running it; the DSL calls are plain Kafka Streams. **Defining the bean.** Typical shape: ```java @Bean public KStream<String, String> pipeline(StreamsBuilder builder) { KStream<String, String> s = builder.stream("input"); s.mapValues(String::toUpperCase).to("output"); return s; } ``` Returning the `KStream` is a convention; the important side effect is that you mutated the injected `builder`. A `void` method that just adds nodes works identically. **Who builds and starts it — the lifecycle.** `StreamsBuilderFactoryBean` implements `SmartLifecycle`. After the context refreshes and all topology beans have contributed to the builder, the factory bean calls `builder.build()` to produce the `Topology`, constructs a `KafkaStreams` object with it plus the merged `StreamsConfig`, and calls `start()`. You never manage this by hand. `autoStartup` (settable on the factory bean) defaults to `true`; set it `false` to defer start and control it programmatically via `factoryBean.start()` / `.stop()`. **Accessing the running client.** Inject the `StreamsBuilderFactoryBean` (by type or the `defaultKafkaStreamsBuilder` name) and call `getKafkaStreams()` to reach the live `KafkaStreams` — e.g. to query state or read `state()`. It returns `null` before start. **Serdes.** Records are bytes; `builder.stream("topic")` uses the default key/value Serde from config (`StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG` / `DEFAULT_VALUE_SERDE_CLASS_CONFIG`) unless you pass `Consumed.with(keySerde, valueSerde)` explicitly. Serde mismatches are a classic runtime failure. **Gotchas.** (1) Don't instantiate `KafkaStreams` yourself — you'll end up with two clients on the same `application.id` and rebalancing chaos. (2) All topology beans share one builder/`application.id`; splitting into truly independent apps requires separate named `StreamsBuilderFactoryBean`s, each with its own config. (3) If a topology bean throws while building, the factory bean fails to start — surfaced at context refresh. (4) The topology is fixed at build time; you can't add nodes after `start()`. **When to use.** Use topology beans for declarative stateful processing (enrichment, joins, aggregations, windowing) that would be awkward and error-prone to hand-roll with plain listeners.