apache/kafka · warning · org.apache.kafka.common.errors.TimeoutException

Timeout of {}ms expired before the position for partition {}

Error message

Timeout of {}ms expired before the position for partition {} could be determined

What it means

TimeoutException thrown by AsyncKafkaConsumer.position(TopicPartition, Duration) when the retry loop expires (timer.notExpired() returns false) before subscriptions.validPosition(partition) returns a non-null FetchPosition. position blocks updating fetch positions until either a valid position is materialized or the supplied timeout elapses; the message names the partition and the elapsed timeout in ms.

Source

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

    @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();
        }
    }

    @Override
    public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions) {
        return committed(partitions, defaultApiTimeoutMs);
    }

    @Override
    public Map<TopicPartition, OffsetAndMetadata> committed(final Set<TopicPartition> partitions,
                                                            final Duration timeout) {
        acquireAndEnsureOpen();
        long start = time.nanoseconds();
        try {
            throwIfGroupIdNotDefined();

View on GitHub (pinned to c31c9215e1)

Solutions

1. Increase the timeout passed to position(partition, Duration.ofSeconds(N)) — give the broker time to resolve the offset.
2. Call consumer.poll(...) once before position so the assignment is materialized and the ListOffsets request has completed.
3. Raise default.api.timeout.ms / request.timeout.ms if the broker or network is consistently slow.
4. Investigate broker health / connectivity (bootstrap reachability, GC stalls, leader election) if timeouts persist.

Example fix

// before
consumer.assign(List.of(tp));
long pos = consumer.position(tp, Duration.ofMillis(50)); // often too tight

// after
consumer.assign(List.of(tp));
consumer.poll(Duration.ofMillis(100)); // materialize the position
long pos = consumer.position(tp, Duration.ofSeconds(5));

// No pre-check can fully prevent a timeout (it depends on broker responsiveness),
// but you can size the timeout to your SLA and sanity-check the inputs.
static long safePosition(Consumer<?, ?> c, TopicPartition tp, Duration timeout) {
    if (timeout == null || timeout.isNegative() || timeout.isZero())
        throw new IllegalArgumentException("position timeout must be positive, got " + timeout);
    if (!c.assignment().contains(tp))
        throw new IllegalStateException(tp + " is not assigned; cannot determine position");
    return c.position(tp, timeout);
}

// Heuristic: pick a timeout >= 2x your typical broker round-trip + rebalance time,
// and never smaller than request.timeout.ms.

// TimeoutException from position() is retriable if the underlying cause is broker
// latency or an in-flight rebalance. Use bounded retries with backoff.
Duration[] backoff = { Duration.ofMillis(100), Duration.ofMillis(500), Duration.ofSeconds(1) };
long offset = -1;
for (int attempt = 0; attempt <= backoff.length; attempt++) {
    try {
        offset = consumer.position(tp, Duration.ofSeconds(10));
        break;
    } catch (org.apache.kafka.common.errors.TimeoutException e) {
        if (attempt == backoff.length)
            throw new IllegalStateException("Could not determine position for " + tp + " after retries", e);
        log.warn("position({}) timed out (attempt {}); backing off {}", tp, attempt + 1, backoff[attempt]);
        try { Thread.sleep(backoff[attempt].toMillis()); } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
            throw new org.apache.kafka.common.errors.InterruptedException(ie);
        }
        // Triggering a poll often forces position resolution post-rebalance:
        try { consumer.poll(Duration.ZERO); } catch (Exception ignored) {}
    }
}

When it happens

Trigger: Calling consumer.position(tp, timeout) with a short timeout before the consumer has resolved a fetch position for tp (e.g. immediately after assign/subscribe, before any poll has triggered a ListOffsets round-trip). Also triggered when the broker is slow to respond to ListOffsets, when the network is unhealthy, or when wakeups interrupt the loop.

Common situations: Tight timeout in tests calling position right after assign; broker under load or recovering; client disconnected/reconnecting during the call; auto.offset.reset=none combined with no committed offset forcing the consumer to wait; cold-start scenarios where the first position resolution requires multiple round-trips.

Related errors

• Timeout of {}ms expired before the last committed offset for
• Failed to get offsets by times in {}ms
• Operation timed out before completion
• Timeout of {}ms expired before successfully committing offse
• Timeout of {}ms expired before the position for partition {}

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