apache/kafka · error · IllegalStateException

You can only check the position for partitions assigned to t

Error message

You can only check the position for partitions assigned to this consumer.

What it means

Thrown by KafkaConsumer.position(TopicPartition, Duration) when the requested partition is not in the consumer's current assignment (subscriptions.isAssigned returns false). The consumer cannot report a fetch position for a partition it is not consuming, so this IllegalStateException signals a logic error in the caller. It occurs before any network call.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java:877

        try {
            Collection<TopicPartition> parts = partitions.isEmpty() ? this.subscriptions.assignedPartitions() : partitions;
            subscriptions.requestOffsetReset(parts, AutoOffsetResetStrategy.LATEST);
        } finally {
            release();
        }
    }

    @Override
    public long position(TopicPartition partition) {
        return position(partition, Duration.ofMillis(defaultApiTimeoutMs));
    }

    @Override
    public long position(TopicPartition partition, final Duration timeout) {
        acquireAndEnsureOpen();
        try {
            if (!this.subscriptions.isAssigned(partition))
                throw new IllegalStateException("You can only check the position for partitions assigned to this consumer.");

            Timer timer = time.timer(timeout);
            do {
                SubscriptionState.FetchPosition position = this.subscriptions.validPosition(partition);
                if (position != null)
                    return position.offset;

                updateFetchPositions(timer);
                client.poll(timer);
            } while (timer.notExpired());

            throw new TimeoutException("Timeout of " + timeout.toMillis() + "ms expired before the position " +
                    "for partition " + partition + " could be determined");
        } finally {
            release();
        }
    }

View on GitHub (pinned to c31c9215e1)

Solutions

Only call position() for partitions returned by consumer.assignment(); verify membership first.
Ensure at least one poll() has completed so assignment is populated before querying position.
Handle ConsumerRebalanceListener to refresh your cached partition set, then guard position() calls against the live assignment.

Example fix

// before
long pos = consumer.position(new TopicPartition("orders", 0));

// after
TopicPartition tp = new TopicPartition("orders", 0);
if (consumer.assignment().contains(tp)) {
    long pos = consumer.position(tp);
}

Defensive patterns

Strategy: validation

Validate before calling

// Only query position() for partitions currently in the assignment:
Set<TopicPartition> assigned = consumer.assignment();
if (assigned.contains(tp)) {
    return consumer.position(tp);
}
throw new IllegalStateException("Partition " + tp + " is not assigned; current: " + assigned);

Type guard

// Narrow a TopicPartition to the 'assigned' subset before use:
static Optional<TopicPartition> ifAssigned(Consumer<?,?> c, TopicPartition tp) {
    return c.assignment().contains(tp) ? Optional.of(tp) : Optional.empty();
}
// Usage: ifAssigned(consumer, tp).ifPresent(consumer::position);

Try / catch

// Recovery: drop the unassigned partition and continue, since position() is undefined for it:
try {
    pos = consumer.position(tp);
} catch (IllegalStateException e) {
    if (e.getMessage().contains("assigned to this consumer")) {
        log.warn("Skipping unassigned {}", tp);
        continue;
    }
    throw e;
}

Prevention

Treat consumer.assignment() as the single source of truth inside a ConsumerRebalanceListener.onAssign/onRevoke boundary; never cache partition sets across rebalances.
Re-derive the partition list right before calling position()/seek()/pause() rather than reusing one captured earlier.
In manual-assignment mode, confirm partition.exists(partition) and the assign() call succeeded before any positional query.

When it happens

Trigger: Calling position(tp) for a TopicPartition obtained from partitionsFor() or listTopics() instead of from assignment(); calling position on a partition whose assignment was revoked after a rebalance; mixing manual assignment and subscription and querying the wrong set.

Common situations: Calling position() immediately after subscribe() before poll() has triggered assignment; reading partition list from topic metadata instead of consumer.assignment(); bug exposed after rebalances that shrink the assignment; threads sharing a partition set without syncing with assignment updates.

Related errors

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