Kafka: Consumer Is Not Subscribed to Any Topics — Fix

You wired up a KafkaConsumer, called poll(), and got a stack trace instead of records. The consumer never even talked to the broker — this one fails client-side, before a single network round-trip.

1. The Error

The exact message, thrown straight out of poll():

Exception in thread "main" java.lang.IllegalStateException: Consumer is not subscribed to any topics or assigned any partitions
	at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedPartitions(SubscriptionState.java:...)
	at org.apache.kafka.clients.consumer.KafkaConsumer.updateAssignmentMetadataIfNeeded(KafkaConsumer.java:...)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:...)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:...)
	at com.example.MyConsumer.main(MyConsumer.java:...)

There are two close cousins from the same subscribe/assign state machine, and they get pasted into search bars just as often:

java.lang.IllegalStateException: Subscription to topics, partitions and pattern are mutually exclusive

java.lang.IllegalStateException: You can only check the position for partitions assigned to this consumer.

This is kafka-clients behavior across the 2.x, 3.x, and 4.0 lines (the ClassicKafkaConsumer and, since 3.7, the AsyncKafkaConsumer under KIP-848 both enforce the same contract). It's client-side and broker-agnostic — Docker, managed Kafka, or a Testcontainers cluster, the message is identical because the check runs before any request leaves the JVM. It's not a subclass of RetriableException; retrying does nothing.

2. How to Reproduce It

You don't need a running cluster to trigger the classic case — the throw happens on the client before the coordinator is ever contacted. But to make the reproduction realistic, spin up a single broker.

# docker-compose.yml — single KRaft broker, Kafka 4.0
services:
  kafka:
    image: apache/kafka:4.0.0
    container_name: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

docker compose up -d
docker exec -it kafka /opt/kafka/bin/kafka-topics.sh \
  --create --topic orders --partitions 3 --replication-factor 1 \
  --bootstrap-server localhost:9092

The consumer that reproduces the classic throw — note there is no subscribe() and no assign():

// kafka-clients 4.0.0
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "orders-app");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);

try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
    // BUG: never subscribed, never assigned
    while (true) {
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
        records.forEach(r -> System.out.println(r.value()));
    }
}

Three more one-line variants, each throwing at poll() or the failing call:

// Variant A — subscribe to an EMPTY collection. This is treated as unsubscribe().
consumer.subscribe(Collections.emptyList());          // clears any subscription
consumer.poll(Duration.ofMillis(500));                // -> IllegalStateException

// Variant B — mixing the two assignment APIs (mutually exclusive)
consumer.subscribe(List.of("orders"));
consumer.assign(List.of(new TopicPartition("orders", 0)));
// -> IllegalStateException: Subscription to topics, partitions and pattern are mutually exclusive

// Variant C — seek/position before an assignment exists
consumer.subscribe(List.of("orders"));
consumer.position(new TopicPartition("orders", 0));   // no poll() yet -> assignment is empty
// -> IllegalStateException: You can only check the position for partitions assigned to this consumer.

3. Why It Happens — Surface Level

KafkaConsumer has two, and only two, ways to get partitions: group subscription (subscribe(), dynamic assignment via the group coordinator) or manual assignment (assign(), you name the exact TopicPartitions). If you call neither — or you call subscribe() with an empty collection, which the client interprets as "unsubscribe" — then when poll() runs its pre-flight check it finds no subscription and no manual assignment, and it refuses to proceed.

It's a fail-fast guard. Without it, poll() would block forever fetching from nowhere, and you'd be staring at a silent hang instead of a stack trace. The exception is the client telling you the object is misconfigured, not that the broker is down.

4. Why It Happens — Under the Hood

Every consumer holds a SubscriptionState, a small state machine with a SubscriptionType field that starts at NONE. The first call that sets partitions locks the type:

• subscribe(Collection<String>) → AUTO_TOPICS
• subscribe(Pattern) → AUTO_PATTERN
• assign(Collection<TopicPartition>) → USER_ASSIGNED

Once the type is anything but NONE, SubscriptionState enforces it: try to assign() while in AUTO_TOPICS, or subscribe() while in USER_ASSIGNED, and you get "Subscription to topics, partitions and pattern are mutually exclusive." The two models are incompatible on purpose — group-managed assignment and manual assignment can't coexist on one consumer, because the coordinator would fight your hand-picked partitions.

At the top of poll(), updateAssignmentMetadataIfNeeded() calls subscriptions.hasNoSubscriptionOrUserAssignment(). If the type is still NONE, it throws "Consumer is not subscribed to any topics or assigned any partitions." This is the guard firing before any Fetch request is built.

The critical subtlety that trips people up: subscribe() is lazy. It records intent (AUTO_TOPICS + the topic set) but does not fetch a partition assignment. The actual assignment materializes only inside poll(), when the consumer joins the group (JoinGroup → SyncGroup) and the coordinator hands back partitions. That's why position(), seek(), seekToBeginning(), pause(), and friends throw "You can only check the position for partitions assigned to this consumer" if you call them after subscribe() but before the first successful poll() — the subscription exists, but the assignment is still empty. (assign(), by contrast, is eager: partitions are set the instant you call it, so seeking right after assign() is fine.)

So the same IllegalStateException class covers three distinct states: no assignment model chosen at all, two models chosen at once, and an assignment model chosen but not yet realized.

5. The Fix

Pick one assignment model and use it correctly.

Group subscription (the common case) — subscribe to a non-empty topic list before the poll loop:

 try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
+    consumer.subscribe(List.of("orders"));
     while (true) {
         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
         records.forEach(r -> System.out.println(r.value()));
     }
 }

Manual assignment (replay tools, partition pinning, no group) — name the partitions explicitly and skip group.id entirely if you're not committing offsets:

+    TopicPartition p0 = new TopicPartition("orders", 0);
+    consumer.assign(List.of(p0));
+    consumer.seekToBeginning(List.of(p0));   // legal: assign() is eager
     while (true) {
         ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
         records.forEach(r -> System.out.println(r.value()));
     }

Fix for Variant A (empty collection) — guard against building the topic list dynamically from config and getting an empty result:

-List<String> topics = config.getTopics();     // returns [] when the property is unset
-consumer.subscribe(topics);
+List<String> topics = config.getTopics();
+if (topics.isEmpty()) {
+    throw new IllegalStateException("No topics configured for consumer group " + groupId);
+}
+consumer.subscribe(topics);

An empty collection isn't an error to the client — subscribe(emptyList()) is documented to be equivalent to unsubscribe(). That's the trap: a misread config file silently disarms the consumer, and you only find out at poll().

Fix for Variant C (seek before assignment) — move the seek into the rebalance callback, where the assignment is guaranteed to exist:

consumer.subscribe(List.of("orders"), new ConsumerRebalanceListener() {
    @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) { }
    @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
        consumer.seekToBeginning(partitions);   // assignment is live here
    }
});

6. Best Practices & The Better Design

Most real-world hits are not "I forgot to subscribe." They're structural: a shared consumer factory whose topic list comes from a property that wasn't set in one environment, a startup ordering bug where poll() runs on a background thread before subscribe() on the main thread, or code that tries to seek() in the constructor.

Subscribe once, immediately before the loop, and never rebuild the subscription per poll. The subscription is sticky state; re-subscribing on every iteration churns the group into needless rebalances.

Don't hand-roll consumers for production. In Spring for Apache Kafka the container owns the entire subscribe/assign/poll lifecycle, and you declare topics on the annotation — there's no code path where you can forget to subscribe:

@KafkaListener(topics = "orders", groupId = "orders-app")
public void onOrder(Order order) {
    process(order);
}

If you need to seek at startup with Spring, don't touch the consumer directly — implement ConsumerSeekAware and use the callback delivered after assignment:

@Component
class OrdersListener extends AbstractConsumerSeekAware {
    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback cb) {
        assignments.keySet().forEach(tp -> cb.seekToBeginning(tp.topic(), tp.partition()));
    }

    @KafkaListener(topics = "orders", groupId = "orders-app")
    void onOrder(Order order) { process(order); }
}

Choose the assignment model deliberately. Use subscribe() when you want the group to balance partitions across instances and handle failover — that's 95% of services. Use assign() only for tools that must own specific partitions with no group coordination: a replay/backfill job, a partition-level debugger, or an external-offset-store design. Never mix them on one consumer, and never share one KafkaConsumer instance between the two styles.

7. How to Prevent It Long-Term

Fail fast on empty config. Validate the topic list at startup and refuse to boot with an empty subscription rather than discovering it at the first poll(). A @Validated @NotEmpty List<String> topics on your config properties catches it before the context finishes loading.

Integration test the subscribe path. A unit test that mocks the consumer won't catch this — the guard lives in the real client. Use Testcontainers to stand up a broker, produce one record, and assert your consumer actually receives it. That single round-trip proves the subscription is wired, the group joins, and the assignment materializes:

@Test
void consumerReceivesRecord() {
    // produce one record to "orders", then:
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
    assertThat(records.count()).isEqualTo(1);
}

Lint for the mixing bug. An ArchUnit or Checkstyle rule that flags any class calling both subscribe( and assign( on the same KafkaConsumer reference catches Variant B before review.

Alert on assigned-partitions = 0. Export the consumer's assigned-partitions metric. A running consumer that never reaches a non-zero assignment — with lag climbing on the topic — is either mis-subscribed or stuck outside the group, and both show up as this flat-at-zero signal.

8. Key Takeaways

• The message is literal. poll() found neither a subscription nor a manual assignment. It's a client-side fail-fast guard, thrown before any broker request — not a network or broker problem, and not retriable.
• subscribe(emptyList()) == unsubscribe(). A topic list built from unset config silently disarms the consumer; validate it at startup.
• subscribe() is lazy, assign() is eager. Seeking or checking position() after subscribe() but before the first poll() throws "You can only check the position for partitions assigned to this consumer" — do it in onPartitionsAssigned.
• The two models are mutually exclusive. Group subscription and manual assignment can't coexist on one consumer; "Subscription to topics, partitions and pattern are mutually exclusive" means you called both.
• Let a framework own the lifecycle. Spring's @KafkaListener removes every code path where you can forget to subscribe; reserve raw assign() for replay tools with no group.