What causes "Consumer is not subscribed to any topics." in apache/kafka?

Ensure consumer.subscribe(Collections.singletonList(topic)) is called before the first poll().

apache/kafka · error · IllegalStateException

Consumer is not subscribed to any topics.

Error message

Consumer is not subscribed to any topics.

What it means

Thrown by ShareConsumerImpl.poll (line 615) when subscriptions.hasNoSubscriptionOrUserAssignment() returns true at poll time. The share consumer requires either an active topic subscription or a user assignment before records can be fetched, mirroring the contract of the classic KafkaConsumer. Without a subscription the broker has no share group assignment to honor, so polling is meaningless and the client fails fast rather than blocking forever.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:615

        acquireAndEnsureOpen();
        try {
            // Throw any errors notified by the background thread
            processBackgroundEvents();

            // Handle any completed acknowledgements for which we already have the responses
            handleCompletedAcknowledgements();

            // If using implicit acknowledgement, acknowledge the previously fetched records
            acknowledgeBatchIfImplicitAcknowledgement();

            // If using explicit acknowledgement, make sure all in-flight records have been acknowledged
            ensureInFlightAcknowledgedIfExplicitAcknowledgement();

            kafkaShareConsumerMetrics.recordPollStart(timer.currentTimeMs());

            if (subscriptions.hasNoSubscriptionOrUserAssignment()) {
                throw new IllegalStateException("Consumer is not subscribed to any topics.");
            }

            shouldSendShareFetchEvent = true;

            // This distinguishes the first pass of the inner do/while loop from subsequent passes for the
            // in-flight poll event logic.
            boolean firstPass = true;

            do {
                // We must not allow wake-ups between polling for fetches and returning the records.
                // A wake-up between returned fetches and returning records would lead to never
                // returning the records in the fetches. Thus, we trigger a possible wake-up before we poll fetches.
                wakeupTrigger.maybeTriggerWakeup();

                // Make sure the network thread can tell the application is actively polling
                checkInFlightPoll(timer, firstPass);
                firstPass = false;
                final ShareFetch<K, V> fetch = pollForFetches(timer);

View on GitHub (pinned to c31c9215e1)

Solutions

Ensure consumer.subscribe(Collections.singletonList(topic)) is called before the first poll().
If you intentionally unsubscribed, do not call poll() again until you re-subscribe or assign partitions.
Guard the poll loop with a check on whether a subscription/assignment exists, or initialize the subscription in the constructor/@PostConstruct of the owning component.
Verify the subscribe() call is not inside a try block that silently swallowed an earlier exception, leaving the consumer un-subscribed.

Example fix

// before
try (var consumer = new KafkaShareConsumer<String,String>(props)) {
    ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));
}

// after
try (var consumer = new KafkaShareConsumer<String,String>(props)) {
    consumer.subscribe(Collections.singletonList("orders"));
    ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(1000));
}

Defensive patterns

Strategy: validation

Validate before calling

// Before calling poll(), ensure the consumer has a subscription or assignment.
java.util.Set<String> sub = consumer.subscription();
java.util.Set<org.apache.kafka.common.TopicPartition> asn = consumer.assignment();
if ((sub == null || sub.isEmpty()) && (asn == null || asn.isEmpty())) {
    throw new IllegalStateException("Cannot poll: consumer has no subscription and no assignment");
}

Type guard

null

Try / catch

try {
    org.apache.kafka.clients.consumer.ConsumerRecords<K,V> records = consumer.poll(java.time.Duration.ofMillis(500));
} catch (IllegalStateException e) {
    if (e.getMessage() != null && e.getMessage().contains("not subscribed")) {
        // ensure subscription is established before retrying
        consumer.subscribe(java.util.List.of("my-topic"));
    } else {
        throw e;
    }
}

Prevention

Always call consumer.subscribe(Collections) or assign(partitions) before the first poll().
Treat subscription as a precondition in a wrapper method that asserts a non-empty subscription set.
If you unsubscribe at runtime, gate subsequent poll() calls behind a state flag.

When it happens

Trigger: Calling consumer.poll(...) on a ShareConsumer (KafkaShareConsumer) before calling consumer.subscribe(...) or consumer.assign(...). Also triggered after consumer.unsubscribe() is invoked and then poll() is called without re-subscribing.

Common situations: App boot sequences where poll() runs in a loop started before the subscribe call completes; refactors that move subscribe() into a conditional branch that was skipped; test harnesses that construct the consumer and immediately poll; misconfigured dependency-injection where the subscribe step was wired to a different bean than the poll loop.

Related errors

AI-assisted analysis of apache/kafka@c31c9215e1 (2026-08-03). Data as JSON: /data/errors/18705fa03669b07c.json. Report an issue: GitHub.