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

Timeout of {}ms expired before the last committed offset for

Error message

Timeout of {}ms expired before the last committed offset for partitions {} could be determined. Try tuning default.api.timeout.ms larger to relax the threshold.

What it means

TimeoutException thrown by AsyncKafkaConsumer.committed(Set<TopicPartition>, Duration) when applicationEventHandler.addAndGet(FetchCommittedOffsetsEvent) raises a TimeoutException before the group coordinator returns the committed offsets. The wrapper message names the timeout (ms), the requested partitions, and explicitly suggests raising default.api.timeout.ms. committed() requires a valid group.id (throwIfGroupIdNotDefined runs first) and an empty partition set short-circuits.

Source

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

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

            final FetchCommittedOffsetsEvent event = new FetchCommittedOffsetsEvent(
                partitions,
                calculateDeadlineMs(time, timeout));
            wakeupTrigger.setActiveTask(event.future());
            try {
                return applicationEventHandler.addAndGet(event);
            } catch (TimeoutException e) {
                throw new TimeoutException("Timeout of " + timeout.toMillis() + "ms expired before the last " +
                    "committed offset for partitions " + partitions + " could be determined. Try tuning " +
                    ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG + " larger to relax the threshold.");
            } finally {
                wakeupTrigger.clearTask();
            }
        } finally {
            kafkaConsumerMetrics.recordCommitted(time.nanoseconds() - start);
            release();
        }
    }

    private void throwIfGroupIdNotDefined() {
        if (groupMetadata.get().isEmpty()) {
            throw new InvalidGroupIdException("To use the group management or offset commit APIs, you must " +
                "provide a valid " + ConsumerConfig.GROUP_ID_CONFIG + " in the consumer configuration.");
        }
    }

View on GitHub (pinned to c31c9215e1)

Solutions

1. Raise default.api.timeout.ms (the message explicitly recommends this) or pass a larger Duration to committed(partitions, timeout).
2. Ensure the consumer has joined the group (poll at least once) before calling committed, so the coordinator context is ready.
3. Check coordinator/broker health and connectivity (bootstrap servers, security, GC), and retry with backoff.
4. Reduce the partition set passed in if you only need a subset, to lower round-trip cost.

Example fix

// before
Map<TopicPartition, OffsetAndMetadata> c =
    consumer.committed(allPartitions, Duration.ofMillis(100));

// after
props.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 60000);
// ... later:
Map<TopicPartition, OffsetAndMetadata> c =
    consumer.committed(allPartitions, Duration.ofSeconds(30));

// committed() requires group.id and a non-empty partition set; check both, and
// size the timeout to your environment.
static Map<TopicPartition, OffsetAndMetadata> safeCommitted(
        Consumer<?, ?> c, Set<TopicPartition> parts, Duration timeout) {
    if (c.groupMetadata() == null)
        throw new IllegalStateException("group.id not set; committed() is unavailable");
    if (parts == null || parts.isEmpty())
        return java.util.Collections.emptyMap();
    if (timeout == null || timeout.isNegative() || timeout.isZero())
        throw new IllegalArgumentException("committed timeout must be positive, got " + timeout);
    return c.committed(parts, timeout);
}

// The error message itself suggests tuning default.api.timeout.ms; pre-flight by
// ensuring the property is set generously relative to request.timeout.ms:
//   default.api.timeout.ms  >=  request.timeout.ms + retry.backoff.ms * (retries)
// A typical safe starting point: request=30s, default.api=60s.

// TimeoutException from committed() is retriable when the broker is reachable.
// Retry with backoff; escalate by widening default.api.timeout.ms if it persists.
int maxAttempts = 3;
long[] backoffMs = { 200, 1000, 5000 };
Map<TopicPartition, OffsetAndMetadata> committed = null;
for (int attempt = 0; ; attempt++) {
    try {
        committed = consumer.committed(parts, Duration.ofSeconds(30));
        break;
    } catch (org.apache.kafka.common.errors.TimeoutException e) {
        if (attempt >= maxAttempts) {
            log.error("committed() timed out for {} after {} attempts; " +
                      "consider raising default.api.timeout.ms and check group coordinator health",
                      parts, attempt);
            throw e;
        }
        log.warn("committed({}) timed out (attempt {}/{}); backing off {}ms",
                 parts, attempt + 1, maxAttempts, backoffMs[attempt]);
        try { Thread.sleep(backoffMs[attempt]); } catch (InterruptedException ie) {
            Thread.currentThread().interrupt();
            throw new org.apache.kafka.common.errors.InterruptedException(ie);
        }
    }
}

When it happens

Trigger: Calling consumer.committed(partitions, timeout) with a timeout too short for the group coordinator to respond; coordinator unavailable or still loading; consumer not yet joined to the group; broker/network latency exceeding the supplied Duration. The catch re-wraps the inner TimeoutException with the actionable hint.

Common situations: Default api timeout overridden too low in tight tests; coordinator rebalancing or just-elected; client connecting to an unreachable/down broker; large partition set with slow offset fetch; running committed() immediately after construction before group join completes; cross-AZ/region latency; running with auto-commit disabled and checking committed offsets in a startup probe.

Related errors

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

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