Kafka Streams MissingSourceTopicException: Fix It Fast

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:

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:

org.apache.kafka.streams.errors.MissingSourceTopicException: Missing source topics: [orders]

Followed shortly by the client shutting itself down:

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.

# 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

// 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<String, String> 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:

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:

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).

- // relying on the topic "just being there"
- KStream<String, String> orders = builder.stream("orders");
+ // provision it explicitly, before the Streams app starts

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.

// 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.

- props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app");
- KStream<String, String> orders = builder.stream("order");     // singular typo
+ props.put(StreamsConfig.APPLICATION_ID_CONFIG, "orders-app-prod");
+ KStream<String, String> orders = builder.stream("orders");    // matches producer

For pattern subscriptions, confirm the regex actually matches live topics:

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:

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:

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:

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<String> present = admin.listTopics().names().get();
    Set<String> required = Set.of("orders", "customers");
    Set<String> 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.