apache/kafka · error · java.lang.IllegalStateException

Consumer is not subscribed to any topics or assigned any par

Error message

Consumer is not subscribed to any topics or assigned any partitions

What it means

IllegalStateException thrown by AsyncKafkaConsumer.poll at the top of the method when subscriptions.hasNoSubscriptionOrUserAssignment() is true. poll() cannot return records without an active source of partitions, so calling it before subscribe() or assign() is treated as a usage bug. This is a fast-fail guard rather than returning empty records indefinitely, surfacing the wiring mistake at the call site.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:938

     *             partitions to consume from or an unexpected error occurred
     * @throws org.apache.kafka.clients.consumer.OffsetOutOfRangeException if the fetch position of the consumer is
     *             out of range and no offset reset policy is configured.
     * @throws org.apache.kafka.common.errors.TopicAuthorizationException if the consumer is not authorized to read
     *             from a partition
     * @throws org.apache.kafka.common.errors.SerializationException if the fetched records cannot be deserialized
     * @throws org.apache.kafka.common.errors.UnsupportedAssignorException if the `group.remote.assignor` configuration
     *             is set to an assignor that is not available on the broker.
     */
    @Override
    public ConsumerRecords<K, V> poll(final Duration timeout) {
        Timer timer = time.timer(timeout);

        acquireAndEnsureOpen();
        try {
            kafkaConsumerMetrics.recordPollStart(timer.currentTimeMs());

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

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

            do {
                // We must not allow wake-ups between polling for fetches and returning the records.
                // If the polled fetches are not empty the consumed position has already been updated in the polling
                // of the fetches. A wakeup 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();

                checkInflightPoll(timer, firstPass);
                firstPass = false;
                final Fetch<K, V> fetch = pollForFetches(timer);
                if (!fetch.isEmpty()) {
                    // before returning the fetched records, we can send off the next round of fetches

View on GitHub (pinned to c31c9215e1)

Solutions

Call consumer.subscribe(topics) or consumer.assign(partitions) before the first consumer.poll(...).
Ensure subscribe is called with a non-empty topic collection and that any regex Pattern actually matches at least one topic.
If you previously called unsubscribe(), re-subscribe or re-assign before polling again.
Guard startup ordering so the polling thread waits on a readiness signal until subscription is established.

Example fix

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

// after
try (var consumer = new KafkaConsumer<String,String>(props)) {
    consumer.subscribe(List.of("orders"));
    ConsumerRecords<String,String> records = consumer.poll(Duration.ofMillis(100));
}

Defensive patterns

Strategy: validation

Validate before calling

// Guard poll() with an explicit readiness check before the first call,
// and again whenever subscription may have been revoked.
static void ensureReadyForPoll(Consumer<?, ?> c) {
    if (c.subscription().isEmpty() && c.assignment().isEmpty()) {
        throw new IllegalStateException(
            "Consumer has neither an active subscription nor a manual assignment; " +
            "call subscribe() or assign() before poll().");
    }
}

// Usage:
//   consumer.subscribe(singleton("orders"));   // or consumer.assign(partitions)
//   ensureReadyForPoll(consumer);
//   ConsumerRecords<K,V> records = consumer.poll(Duration.ofMillis(500));
//
// For the async consumer, prefer checking c.subscription()/c.assignment() over
// keeping your own flag — they reflect rebalances that may have revoked partitions.

Try / catch

// IllegalStateException from poll() indicates a programmer error, not a transient
// condition. Do NOT retry blindly; fix the subscription state.
try {
    ensureReadyForPoll(consumer);                 // cheap pre-check
    records = consumer.poll(Duration.ofMillis(500));
} catch (IllegalStateException e) {
    if (e.getMessage().contains("not subscribed")) {
        log.warn("No subscription/assignment; re-subscribing before next poll cycle");
        consumer.subscribe(topics, listener);     // recover state, then continue
        // skip this poll, try again next loop iteration
    } else {
        throw e;                                  // different IllegalStateException — propagate
    }
}

Prevention

Always pair subscribe() or assign() with the first poll(); never poll in a code path that may have skipped subscription (e.g. an early-return branch).
In a rebalance callback (ConsumerRebalanceListener.onPartitionsRevoked), do not assume partitions remain assigned after the callback returns; re-check before the next poll if your logic depends on them.
Centralize your poll loop in one method so the subscribe/assign precondition is guaranteed at every call site.
Treat 'Consumer is not subscribed' as a bug in your wiring, not a runtime hiccup — log a stack trace and fix the code path.

When it happens

Trigger: Invoking consumer.poll(timeout) before any consumer.subscribe(...) or consumer.assign(...) call. Also possible if subscription was cleared (unsubscribe) and poll is invoked again without re-subscribing, or if the subscribe call silently failed (e.g. an empty topic collection).

Common situations: Application startup ordering bug where poll runs in a background thread before the subscription step completes; refactoring that moves subscribe() into a conditional branch not always taken; subscribe(Collections.emptyList()) followed by poll; consumer reused after unsubscribe() without re-subscribing; race between main thread subscribing and a health-check thread polling.

Related errors

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