apache/kafka · error · java.lang.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

IllegalStateException thrown by AsyncKafkaConsumer.position(TopicPartition, Duration) when the requested partition is not in the consumer's current assignment (subscriptions.isAssigned(partition) is false). position requires an active fetch position, which only exists for assigned partitions; querying a partition you did not subscribe/assign yields no meaningful value. The check runs inside the acquired consumer lock before any retry/timeout loop.

Source

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

                offsetResetStrategy,
                defaultApiTimeoutDeadlineMs())
            );
        } finally {
            release();
        }
    }

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

    @Override
    public long position(TopicPartition partition, Duration timeout) {
        acquireAndEnsureOpen();
        try {
            if (!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 = subscriptions.validPosition(partition);
                if (position != null)
                    return position.offset;

                updateFetchPositions(timer);
                timer.update();
                wakeupTrigger.maybeTriggerWakeup();
            } 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

1. Only call position for partitions returned by consumer.assignment() (or within an onPartitionsAssigned callback).
2. Refresh the assignment after a rebalance before querying positions; cache nothing across rebalances.
3. If the partition is genuinely not assigned, do not call position — reassign or rebalance first.

Example fix

// before
TopicPartition tp = new TopicPartition("orders", 5); // not necessarily assigned
long pos = consumer.position(tp);

// after
TopicPartition tp = new TopicPartition("orders", 5);
if (consumer.assignment().contains(tp)) {
    long pos = consumer.position(tp);
} else {
    // reassign or skip
}

// position(TopicPartition) requires the partition to be currently assigned.
// Check assignment immediately before the call — assignments change on rebalance.
static long safePosition(Consumer<?, ?> c, TopicPartition tp, Duration timeout) {
    Set<TopicPartition> assigned = c.assignment();
    if (!assigned.contains(tp)) {
        throw new IllegalStateException(
            "Cannot query position for " + tp + "; it is not among the currently assigned " +
            "partitions: " + assigned);
    }
    return c.position(tp, timeout);
}

// Usage:
//   long off = safePosition(consumer, tp, Duration.ofSeconds(10));
//
// Note: assignment() reflects the latest rebalance; calling it right before position()
// minimizes the (non-zero) window in which a revocation could still race you.

// Brand a TopicPartition as 'Assigned' so unassigned partitions cannot be passed
// to position() by construction.
public static Set<TopicPartition> assignedSet(Consumer<?, ?> c) {
    return c.assignment(); // already an immutable snapshot
}
// Then accept only partitions proven to be in that set:
static long positionOf(Consumer<?, ?> c, TopicPartition tp, Duration timeout) {
    if (!c.assignment().contains(tp))
        throw new IllegalStateException(tp + " not assigned");
    return c.position(tp, timeout);
}
//
// TypeScript analogue:
//   type Assigned = TopicPartition & { __brand: 'Assigned' };
//   function assigned(c: Consumer): Assigned[] { return c.assignment() as Assigned[]; }
//   function position(c: Consumer, tp: Assigned): number { return c.position(tp); }

// IllegalStateException from position() means the partition isn't assigned right now;
// the correct response is to refresh assignment, not to retry blindly.
try {
    long off = consumer.position(tp, Duration.ofSeconds(10));
} catch (IllegalStateException e) {
    if (e.getMessage().contains("assigned to this consumer")) {
        log.info("{} not currently assigned; refreshing assignment snapshot", tp);
        Set<TopicPartition> current = consumer.assignment();
        // re-evaluate: skip this partition, or wait for the next poll() to trigger rebalance
        continue; // in a per-partition loop
    }
    throw e;
}

When it happens

Trigger: Calling consumer.position(tp) for a TopicPartition that is not part of the consumer's current assignment. Commonly happens when code computes partitions from topic metadata rather than from consumer.assignment(), or when checking position for a partition whose assignment was lost after a rebalance.

Common situations: Calling position on a partition the user assumed was assigned but was reassigned to another consumer in the group after a rebalance; mixing partition sets across multiple consumer instances; using AdminClient-described partitions to call position instead of consumer.assignment(); race between a rebalance callback revoking partitions and a position() call on the same thread.

Related errors

• Missing position for fetchable partition {}
• Timeout of {}ms expired before the position for partition {}
• Topic partitions collection to assign to cannot be null
• Operation timed out before completion
• You can only check the position for partitions assigned to t

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