> ## Content Index > Fetch the complete content index at: https://www.ggorantala.dev/llms.txt > Use this file to discover other available public pages before exploring further. # Kafka Streams MissingSourceTopicException: Fix It Fast - URL: https://www.ggorantala.dev/kafka-streams-missingsourcetopicexception-fix/ - Published: 2026-05-30T13:48:00.000Z - Updated: 2026-08-29T08:23:42.000Z - Description: Kafka Streams dies with MissingSourceTopicException on startup or rebalance. Here's why the assignor flags a missing source topic and how to fix it. - Author: Gopi Gorantala - Tags: kafka-errors, kafka-streams, MissingSourceTopicException, StreamsException, kafka-docker, Java, event-streaming Your Kafka Streams app boots, joins the group, and seconds later the whole client goes to `ERROR` and stops. The stack trace names a topic you *thought* existed. This is `MissingSourceTopicException`, and unlike most Streams failures it is not a state-store, SerDe, or rebalance-timing problem. It is the assignor telling you that a topic your topology reads from is not in cluster metadata — and by design, Streams will not create it for you. This one bites hardest in POCs and fresh environments (topic never provisioned), in shared clusters (typo or wrong `application.id` prefix), and after someone deletes or recreates a topic under a running app. ## 1\. The Error The exact exception, as it reaches your logs and your `StreamsUncaughtExceptionHandler`: ```text org.apache.kafka.streams.errors.MissingSourceTopicException: One or more source topics were missing during rebalance at org.apache.kafka.streams.processor.internals.StreamsRebalanceListener.onAssignment(StreamsRebalanceListener.java:83) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.invokeOnAssignment(ConsumerCoordinator.java:295) at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.onJoinComplete(ConsumerCoordinator.java:388) ... at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:592) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:551) ``` On Kafka 3.8+ (KAFKA-15951) the message includes the topic names, which is what you actually want: ```text org.apache.kafka.streams.errors.MissingSourceTopicException: Missing source topics: [orders] ``` Followed shortly by the client shutting itself down: ```text INFO o.a.k.s.p.internals.StreamThread - stream-thread [orders-app-...] State transition from RUNNING to PENDING_SHUTDOWN ERROR o.apache.kafka.streams.KafkaStreams - stream-client [orders-app] All stream threads have died. The instance will be in error state and should be closed. INFO o.apache.kafka.streams.KafkaStreams - stream-client [orders-app] State transition from REBALANCING to PENDING_ERROR to ERROR ``` - **Class hierarchy:** `MissingSourceTopicException` → `StreamsException` → `KafkaException`. It is *not* retriable and it is *not* a poison-pill record — it fires at assignment time, before any record is processed. - **Where it comes from:** the consumer group *leader* runs the `StreamsPartitionAssignor`, notices a source topic is absent from metadata, and encodes the assignment error `INCOMPLETE_SOURCE_TOPIC_METADATA`. Every `StreamThread` that receives that assignment throws. - **Versions:** behavior introduced in **Kafka 2.7.0** (KAFKA-10355 / KIP-662). Applies to kafka-streams 2.7 through 4.0, KRaft or ZooKeeper, self-managed or Confluent Cloud. Before 2.7, a deleted source topic caused a silent graceful shutdown instead of an exception. ## 2\. How to Reproduce It Minimal KRaft single-broker cluster, no topic pre-created. ```yaml # docker-compose.yml services: kafka: image: apache/kafka:3.9.0 container_name: kafka ports: - "9092:9092" environment: KAFKA_NODE_ID: 1 KAFKA_PROCESS_ROLES: broker,controller KAFKA_LISTENERS: PLAINTEXT://:19092,CONTROLLER://:9093,EXTERNAL://:9092 KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:19092,EXTERNAL://localhost:9092 KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093 KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 # Off on purpose: source topics must be provisioned, not conjured. KAFKA_AUTO_CREATE_TOPICS_ENABLE: "false" CLUSTER_ID: 4L6g3nShT-eMCtK--X86sw ``` ```java // build.gradle: implementation 'org.apache.kafka:kafka-streams:3.9.0' import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.*; import org.apache.kafka.streams.kstream.KStream; import java.util.Properties; public class OrdersApp { public static void main(String[] args) { Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app"); 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.String().getClass()); StreamsBuilder builder = new StreamsBuilder(); KStream orders = builder.stream("orders"); // topic does NOT exist orders.mapValues(v -> v.toUpperCase()).to("orders-upper"); KafkaStreams streams = new KafkaStreams(builder.build(), props); streams.setUncaughtExceptionHandler(e -> { System.err.println("UNCAUGHT: " + e); return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.SHUTDOWN_CLIENT; }); streams.start(); Runtime.getRuntime().addShutdownHook(new Thread(streams::close)); } } ``` Bring up the broker and run the app: ```bash docker compose up -d # Do NOT create the "orders" topic. java -jar orders-app.jar ``` The app joins the group, the assignor can't find `orders`, and within a second the thread throws `MissingSourceTopicException` and the client transitions to `ERROR`. Create the topic and it starts cleanly: ```bash docker exec kafka /opt/kafka/bin/kafka-topics.sh \ --bootstrap-server localhost:19092 \ --create --topic orders --partitions 6 --replication-factor 1 ``` **Environment-specific triggers to note:** a `Pattern` subscription (`builder.stream(Pattern.compile("orders-.*"))`) that matches *zero* topics throws the same thing. So does deleting `orders` while the app runs — the next metadata refresh flips the leader into `INCOMPLETE_SOURCE_TOPIC_METADATA` and every instance dies. ## 3\. Why It Happens — Surface Level Your topology declares one or more **source topics** (every `builder.stream(...)`, `builder.table(...)`, and `Pattern` subscription). At assignment time the group leader resolves those names against the cluster's topic metadata. If any named topic — or every topic a pattern was supposed to match — is absent, the leader cannot compute a partition assignment for it, so it flags the assignment as incomplete and each thread throws. Streams **auto-creates its own internal topics** (repartition and changelog topics) through its embedded `AdminClient`. It deliberately does **not** auto-create *source* topics. Source topics are your input contract; Streams assumes an upstream producer owns them. So "just turn on `auto.create.topics.enable`" is the wrong reflex — Streams resolves source topics via `AdminClient` metadata, and a missing one is treated as a hard error, not a create trigger. ## 4\. Why It Happens — Under the Hood The mechanics are worth understanding because they explain why *every* instance dies, not just the leader. Kafka Streams runs partition assignment through a custom `StreamsPartitionAssignor` embedded in the consumer group protocol. On every rebalance, one member is elected group leader and runs the assignor. The assignor must map source-topic partitions (plus internal repartition/changelog partitions) onto tasks, and tasks onto members. To do that it needs the partition count of every source topic, which it pulls from the cluster metadata (`Cluster.partitionsForTopic`). When a source topic is missing from that metadata, the assignor can't size the task graph. Rather than assign a broken topology, it encodes `AssignorError.INCOMPLETE_SOURCE_TOPIC_METADATA` into the `SubscriptionInfo`/`AssignmentInfo` payload it hands back to every member via `SyncGroup`. Each member's `StreamsRebalanceListener.onAssignment(...)` reads that error code and throws `MissingSourceTopicException` from the poll thread. That's why the failure fans out: the leader made the decision, but *all* members receive the poisoned assignment and throw. The exception is not caught inside the runtime. It bubbles up the `StreamThread` and hits your `StreamsUncaughtExceptionHandler` (KIP-671, Kafka 2.8+). If you never set one, the **default response is `SHUTDOWN_CLIENT`** — so the default posture is: any source topic goes missing, the whole `KafkaStreams` instance shuts down to `ERROR`. That is intentional. KIP-662 changed the pre-2.7 behavior (silent graceful shutdown that looked like a healthy stop) precisely because accidentally deleting a source topic used to fail *quietly*; now it fails *loudly*. One subtlety on secured clusters: metadata is filtered by authorization. If the app principal lacks `DESCRIBE` on the source topic, the broker returns `UNKNOWN_TOPIC_OR_PARTITION` — the topic looks *missing* even though it exists. An ACL gap and a genuinely absent topic produce the identical `MissingSourceTopicException`. Always rule out authorization before assuming the topic isn't there. ## 5\. The Fix There is no config flag that makes a missing source topic appear. The fix is to guarantee the topic exists (and is visible) before the app joins the group. **Fix A — Provision the source topic (the actual cause 90% of the time).** ```diff - // relying on the topic "just being there" - KStream orders = builder.stream("orders"); + // provision it explicitly, before the Streams app starts ``` ```bash kafka-topics.sh --bootstrap-server localhost:19092 \ --create --topic orders --partitions 6 --replication-factor 3 \ --if-not-exists ``` Better: own topics as code so this can't drift. ```java // Provision inputs with AdminClient at deploy time (or via Strimzi/Terraform). try (Admin admin = Admin.create(Map.of("bootstrap.servers", "localhost:9092"))) { admin.createTopics(List.of( new NewTopic("orders", 6, (short) 3) )).all().get(); } catch (ExecutionException e) { if (!(e.getCause() instanceof TopicExistsException)) throw e; // idempotent } ``` **Fix B — Fix the name, prefix, or pattern.** A missing topic is very often a *typo* or an environment mismatch, not a genuinely absent stream. ```diff - props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app"); - KStream orders = builder.stream("order"); // singular typo + props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app-prod"); + KStream orders = builder.stream("orders"); // matches producer ``` For pattern subscriptions, confirm the regex actually matches live topics: ```bash kafka-topics.sh --bootstrap-server localhost:19092 --list | grep -E '^orders-.*' ``` **Fix C — Grant DESCRIBE (secured clusters).** If the topic exists but the app can't see it: ```bash kafka-acls.sh --bootstrap-server localhost:19092 --add \ --allow-principal User:orders-app \ --operation DESCRIBE --operation READ \ --topic orders ``` **Fix D — Choose how to react to *deletion* at runtime.** If a source topic can legitimately disappear (multi-tenant, dynamic topics), decide the response instead of taking the default client shutdown: ```java streams.setUncaughtExceptionHandler(ex -> { if (ex instanceof MissingSourceTopicException) { log.error("Source topic vanished; shutting down for orchestrator restart", ex); return StreamThreadExceptionResponse.SHUTDOWN_CLIENT; // let k8s restart & re-resolve } return StreamThreadExceptionResponse.REPLACE_THREAD; }); ``` Do **not** blanket-`REPLACE_THREAD` on this exception — a genuinely missing topic will just make the replacement thread throw again in a tight loop. ## 6\. Best Practices & The Better Design The class of bug disappears when input topics are a *provisioned contract*, not an assumption. Treat every source topic as infrastructure that exists before any app that reads it. Provision inputs declaratively — Strimzi `KafkaTopic` CRDs, Terraform, or an `AdminClient` step in your deploy pipeline — and set `auto.create.topics.enable=false` on the broker so nothing silently half-creates a zero-partition ghost. Streams should *find* its inputs, never race to create them. The "right way" wiring: ```java Properties props = new Properties(); props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app-" + env); // env in the id props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap); props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 3); // Fail fast, in your control, before joining the group: try (Admin admin = Admin.create(Map.of("bootstrap.servers", bootstrap))) { Set present = admin.listTopics().names().get(); Set required = Set.of("orders", "customers"); Set missing = new HashSet<>(required); missing.removeAll(present); if (!missing.isEmpty()) { throw new IllegalStateException("Refusing to start; missing source topics: " + missing); } } KafkaStreams streams = new KafkaStreams(topology, props); streams.setUncaughtExceptionHandler(new OrdersExceptionHandler()); streams.start(); ``` That preflight check turns a mid-rebalance client death into a clear startup failure with a precise message — the same information, but *before* you've joined a group and destabilized the other instances. Pin `application.id` to the environment so a staging app can never bind to prod topics (or vice-versa), and keep it stable across restarts so you don't strand internal topics under an orphaned prefix. For truly dynamic topic sets, prefer an explicit orchestrator-driven restart (`SHUTDOWN_CLIENT` \+ a supervised container) over in-process thread replacement, so the app re-resolves metadata cleanly on the next boot. ## 7\. How to Prevent It Long-Term Catch it in CI and in monitoring, not in prod logs. - **CI smoke test with Testcontainers:** start a real broker, deliberately *omit* one source topic, and assert the app reaches `ERROR` (or your preflight throws). Then create the topics and assert it reaches `RUNNING`. Mocks never exercise the assignor, so a unit test can't catch this. - **Alert on client state, not just crashes.** Export `KafkaStreams.state()` and page on `ERROR`/`PENDING_ERROR`. Alert on `kafka.streams:type=stream-metrics,...` `failed-stream-threads` climbing and on `alive-stream-threads` dropping to zero. - **Broker-side signal:** watch for a spike in `UNKNOWN_TOPIC_OR_PARTITION` errors and any unexpected topic *deletions* — automation that cleans up `*-changelog`/`*-repartition` topics or reaps "unused" topics is a classic source-topic killer. - **Config conventions:** `auto.create.topics.enable=false` everywhere, source topics owned as code and reviewed in PRs, `application.id` carrying the environment, and a lint/ArchUnit rule that every `builder.stream/table(...)` name resolves to a declared topic in your topic manifest. - **Guard the ACLs:** include `DESCRIBE`/`READ` on all source topics in the same as-code definition as the topics themselves, so a topic and its read grant ship together. ## 8\. Key Takeaways - `MissingSourceTopicException` is an **assignment-time** failure (`INCOMPLETE_SOURCE_TOPIC_METADATA`), not a record or SerDe problem — the leader flags it and **every** instance throws. - Streams auto-creates internal (repartition/changelog) topics but **never source topics** — turning on broker auto-create won't fix it and will hide worse bugs. - Default reaction is `SHUTDOWN_CLIENT` (KIP-671, 2.7+ throws per KIP-662). That's intentional: a vanished input should fail loud, not silently stop. - On secured clusters, a **missing `DESCRIBE` ACL** looks identical to a missing topic — check authorization before assuming the topic is gone. - Provision source topics as code, add a startup preflight `listTopics` check, and pin `application.id` to the environment. Verify with a Testcontainers test that omits a topic on purpose.