apache/kafka · error · OffsetOutOfRangeException

Fetch position {} is out of range for partition {}

Error message

Fetch position {} is out of range for partition {}

What it means

OffsetOutOfRangeException thrown at FetchCollector.java:359 when the broker returns OFFSET_OUT_OF_RANGE for the leader fetch (no preferred replica to clear) and the consumer has no default offset-reset policy (auto.offset.reset=none). The fetch position is older than the log start offset or ahead of the high watermark, and the client cannot recover automatically, so it surfaces the error to the application.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/FetchCollector.java:359

        } else if (error == Errors.OFFSET_OUT_OF_RANGE) {
            Optional<Integer> clearedReplicaId = subscriptions.clearPreferredReadReplica(tp);

            if (clearedReplicaId.isEmpty()) {
                // If there's no preferred replica to clear, we're fetching from the leader so handle this error normally
                SubscriptionState.FetchPosition position = subscriptions.positionOrNull(tp);

                if (position == null || fetchOffset != position.offset) {
                    log.debug("Discarding stale fetch response for partition {} since the fetched offset {} " +
                            "does not match the current offset {} or the partition has been unassigned", tp, fetchOffset, position);
                } else {
                    String errorMessage = "Fetch position " + position + " is out of range for partition " + tp;

                    if (subscriptions.hasDefaultOffsetResetPolicy()) {
                        log.info("{}, resetting offset", errorMessage);
                        subscriptions.requestOffsetResetIfPartitionAssigned(tp);
                    } else {
                        log.info("{}, raising error to the application since no reset policy is configured", errorMessage);
                        throw new OffsetOutOfRangeException(errorMessage,
                                Collections.singletonMap(tp, position.offset));
                    }
                }
            } else {
                log.debug("Unset the preferred read replica {} for partition {} since we got {} when fetching {}",
                        clearedReplicaId.get(), tp, error, fetchOffset);
            }
        } else if (error == Errors.TOPIC_AUTHORIZATION_FAILED) {
            //we log the actual partition and not just the topic to help with ACL propagation issues in large clusters
            log.warn("Not authorized to read from partition {}.", tp);
            throw new TopicAuthorizationException(Collections.singleton(tp.topic()));
        } else if (error == Errors.UNKNOWN_LEADER_EPOCH) {
            log.debug("Received unknown leader epoch error in fetch for partition {}", tp);
        } else if (error == Errors.UNKNOWN_SERVER_ERROR) {
            log.warn("Unknown server error while fetching offset {} for topic-partition {}",
                    fetchOffset, tp);
        } else if (error == Errors.CORRUPT_MESSAGE) {
            throw new KafkaException("Encountered corrupt message when fetching offset "

View on GitHub (pinned to c31c9215e1)

Solutions

  1. Set auto.offset.reset to earliest or latest so the consumer can auto-reset when the position is out of range.
  2. Seek the affected partition to a valid offset: earliest(), latest(), or seekToBeginning/seekToEnd before the next poll.
  3. Increase retention (log.retention.hours / retention.bytes) on the broker so committed offsets remain valid.
  4. Reset the consumer group offsets via kafka-consumer-groups.sh --reset-offsets after data loss/recreation.

Example fix

// before
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");

// after
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// or, in the rebalance listener, reset explicitly:
consumer.seekToBeginning(Collections.singleton(tp));
Defensive patterns

Strategy: try-catch

Validate before calling

// Configure an auto offset reset policy so the client handles out-of-range internally
// instead of throwing OffsetOutOfRangeException at FetchCollector.java:359.
Properties p = new Properties();
p.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // or "latest", "none"
// If you set "none" intentionally (to detect data loss), validate before poll:
Map<TopicPartition, Long> endOffsets = consumer.endOffsets(assigned);
Map<TopicPartition, Long> beginning = consumer.beginningOffsets(assigned);
for (TopicPartition tp : assigned) {
    long pos = consumer.position(tp);
    if (pos < beginning.get(tp) || pos > endOffsets.get(tp)) {
        // position is off-log; seek to a safe offset BEFORE poll()
        consumer.seek(tp, OffsetResetStrategy.EARLIEST.equals(policy) ? beginning.get(tp) : endOffsets.get(tp));
    }
}

Try / catch

try {
    records = consumer.poll(Duration.ofMillis(500));
} catch (org.apache.kafka.common.errors.OffsetOutOfRangeException e) {
    // FetchCollector.java:359: fetch offset off the log AND auto.offset.reset=none
    // (or no default reset policy). Recover explicitly.
    Map<TopicPartition, Long> outOfRange = e.offsetOutOfRangePartitions();
    Map<TopicPartition, Long> beginnings = consumer.beginningOffsets(outOfRange.keySet());
    for (Map.Entry<TopicPartition, Long> entry : outOfRange.entrySet()) {
        TopicPartition tp = entry.getKey();
        long safe = Math.max(beginnings.get(tp), entry.getValue()); // or seek to end
        log.warn("Offset {} out of range for {}; seeking to {}", entry.getValue(), tp, safe);
        consumer.seek(tp, safe);
    }
}

Prevention

When it happens

Trigger: Triggered in handleInitializeErrors when error == Errors.OFFSET_OUT_OF_RANGE, preferredReadReplica is empty, the fetch offset matches the current position, and subscriptions.hasDefaultOffsetResetPolicy() is false. Common after the retained data older than the committed offset has been log-compacted or deleted, or the consumer committed an offset past the partition end.

Common situations: auto.offset.reset=none with committed offsets outside the current log range; retention deleted segments below the committed offset; consumer group reused against a topic whose data was cleared; manual seek() to a position beyond the log end; compacted topic where the committed offset points to a gap.

Related errors

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