How do you inspect a Kafka Streams topology, and why does naming DSL operators and stateful stores matter for topology compatibility?

## Inspecting a topology ```java Topology topology = builder.build(); System.out.println(topology.describe()); ``` `Topology#describe()` returns a **`TopologyDescription`**. Its string form groups the graph into **`Sub-topology: N`** blocks, each listing: - **Source** nodes (the input topics, including internal repartition topics), - **Processor** nodes (with the **state stores** they attach, in `stores: [...]`), - **Sink** nodes (output topics, including internal repartition topics). Disconnected sub-topologies and repartition boundaries are visible here. The community **kafka-streams-viz** / Streams TopologyViz renders this text into a DAG diagram — invaluable for reviewing where repartitions and stores live. ## The naming problem (the principal-level point) Kafka Streams **auto-names** every processor node using a **monotonic index based on its position** in the DSL graph, e.g.: ``` KSTREAM-SOURCE-0000000000 KSTREAM-KEY-SELECT-0000000001 KSTREAM-AGGREGATE-0000000003 ``` Two classes of **internal topics** derive their names from these node names: 1. **Repartition topics**: `<application.id>-<node-name>-repartition` 2. **Changelog topics** (backing state stores): `<application.id>-<store-name>-changelog` ### Why auto-names are fragile The index is assigned by **traversal order**. If you later **insert a new operator**, **reorder branches**, or **upgrade a library version** that changes graph construction, the indices for *downstream* nodes **shift**. That changes the derived **internal topic names**. On the next deployment: - Streams looks for changelog/repartition topics under the **new** names, finds them empty/absent, and **can't restore** the existing state — it may reprocess from scratch or, worse, leave the old topics orphaned holding the real state. - This is a classic cause of broken **rolling upgrades** of stateful Streams apps. ### The fix: name your operators and stores Give **explicit, stable names** so node and internal-topic names don't depend on position: - `Named.as("...")` — on stateless operators (filter, map, peek, branch, etc.) and as an overload on many operators. - `Materialized.as("store-name")` — names a state store (and thus its `-changelog` topic). - `Repartitioned.as("...")` — names an explicit repartition topic. - `Grouped.as("...")` — names the repartition topic created by `groupBy`. - `Joined.as("...")` — names join repartition topics/stores. - `StreamJoined.as("...")` — names the join's state stores. With everything named, you can refactor the DSL (insert filters, reorder stages) and remain **topology-compatible** with state already on disk and in changelog topics. ### Verifying compatibility Because `describe()` is deterministic for a given graph, teams **snapshot the describe() output in a test** and diff it across commits — a changed internal-topic name in the diff flags a breaking, non-rolling-upgradeable change before it ships. ### Related: topology optimization `builder.build(props)` with `topology.optimization=all` (a.k.a. `StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG`) can **reuse the source topic as the changelog** and collapse redundant repartition/through topics — which also changes the graph, so it must be set consistently and is another reason to verify `describe()` across changes. ### Edge cases - Names must be **unique** within the topology; duplicates throw at build time. - Naming the **store** fixes the changelog name but you should also name the **repartition** producers (via Grouped/Joined/Repartitioned) since those are separately derived. - Changing `application.id` changes ALL internal topic names (it's the prefix) — that's a full reset, distinct from the position-index problem.