apache/kafka · error · NoOffsetForPartitionException

Undefined offset with no reset policy for partitions: ${part

Error message

Undefined offset with no reset policy for partitions: ${partitionsWithNoOffsets}

What it means

Thrown as NoOffsetForPartitionException by resetInitializingPositions when at least one assigned partition has no committed offset (no position) and the consumer's defaultResetStrategy is AutoOffsetResetPolicy.NONE (i.e. auto.offset.reset=none). The client refuses to silently pick earliest/latest and instead surfaces the gap so the application decides. This is the only path that raises NoOffsetForPartitionException; it fires during the position-initialization phase of poll().

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java:882

     *
     * @param initPartitionsToInclude Initializing partitions to include in the reset. Assigned partitions that
     *                                require a positions but are not included in this set won't be reset.
     * @throws NoOffsetForPartitionException If there are partitions assigned that require a position but
     *                                       there is no reset strategy configured.
     */
    public synchronized void resetInitializingPositions(Predicate<TopicPartition> initPartitionsToInclude) {
        final Set<TopicPartition> partitionsWithNoOffsets = new HashSet<>();
        assignment.forEach((tp, partitionState) -> {
            if (partitionState.shouldInitialize() && initPartitionsToInclude.test(tp)) {
                if (defaultResetStrategy == AutoOffsetResetStrategy.NONE)
                    partitionsWithNoOffsets.add(tp);
                else
                    requestOffsetReset(tp);
            }
        });

        if (!partitionsWithNoOffsets.isEmpty())
            throw new NoOffsetForPartitionException(partitionsWithNoOffsets);
    }

    public synchronized void resetInitializingPositions() {
        resetInitializingPositions(tp -> true);
    }

    public synchronized Set<TopicPartition> partitionsNeedingReset(long nowMs) {
        return collectPartitions(state -> state.awaitingReset() && !state.awaitingRetryBackoff(nowMs));
    }

    public synchronized Map<TopicPartition, FetchPosition> partitionsNeedingValidation(long nowMs) {
        Map<TopicPartition, FetchPosition> result = new HashMap<>();

        assignment.forEach((tp, tps) -> {
            if (tps.awaitingValidation() && !tps.awaitingRetryBackoff(nowMs) && tps.position != null) {
                result.put(tp, tps.position);
            }
        });

View on GitHub (pinned to c31c9215e1)

Solutions

Set auto.offset.reset to earliest or latest (whichever matches your data semantics) so the client can deterministically reset.
If 'none' is intentional for safety, catch NoOffsetForPartitionException, inspect partitionsWithNoOffsets, and explicitly consumer.seek() each partition to a chosen offset, then resume polling.
Use kafka-consumer-groups --reset-offsets to seed committed offsets before starting consumers if you need a one-time reset.
Raise offsets.retention.minutes on the broker if offsets are being aged out faster than consumer downtime.

Example fix

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

// after
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// or handle explicitly:
try {
    consumer.poll(Duration.ofMillis(500));
} catch (NoOffsetForPartitionException e) {
    for (TopicPartition tp : e.partitions()) consumer.seek(tp, 0L);
}

Defensive patterns

Strategy: try-catch

Validate before calling

// Avoid auto.offset.reset=none unless you explicitly want this failure.
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // or "latest"
// If you keep NONE, inspect committed offsets before the first poll loop:
Map<TopicPartition, OffsetAndMetadata> committed = consumer.committed(consumer.assignment());
Set<TopicPartition> missing = consumer.assignment().stream()
    .filter(tp -> committed == null || committed.get(tp) == null)
    .collect(Collectors.toSet());
if (!missing.isEmpty()) {
    consumer.seekToBeginning(missing); // explicit policy decision instead of throwing
}

Type guard

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;

/** Partitions in 'assigned' with no committed offset (would trigger NoOffsetForPartitionException under reset=none). */
static java.util.Set<TopicPartition> partitionsWithoutOffset(
        Map<TopicPartition, OffsetAndMetadata> committed,
        java.util.Set<TopicPartition> assigned) {
    java.util.Set<TopicPartition> out = new java.util.HashSet<>();
    for (TopicPartition tp : assigned) {
        OffsetAndMetadata o = committed == null ? null : committed.get(tp);
        if (o == null || o.offset() < 0) out.add(tp);
    }
    return out;
}

Try / catch

try {
    ConsumerRecords<K,V> records = consumer.poll(Duration.ofMillis(1000));
} catch (org.apache.kafka.clients.consumer.NoOffsetForPartitionException e) {
    // Apply an explicit reset instead of relying on the global policy.
    consumer.seekToBeginning(e.partitions());   // or seekToEnd(e.partitions())
    // optionally: commit the chosen offsets so the next restart is deterministic
}

Prevention

Never set auto.offset.reset=none in production unless you wrap poll() in a NoOffsetForPartitionException handler.
Commit offsets before shutting down so the next start has a reset point.
On first deployment, bootstrap offsets via seekToBeginning/seekToEnd or a one-time loader.
Log committed-offset gaps during consumer startup to catch missing-offset state early.

When it happens

Trigger: Configuring auto.offset.reset=none and then polling a group/topic where one or more partitions have no committed offset for this consumer group (new group, expired offsets after offset.retention, partition count change, or __consumer_offsets compaction/TTL). Also triggered after manually deleting a group's offsets via kafka-consumer-groups --delete-offsets.

Common situations: Production default of auto.offset.reset=none for safety, deployed against a brand-new topic; offsets expired because offsets.retention.minutes (default 7 days) lapsed during a consumer outage; partitions added by increasing topic partitions and the new partitions have no committed offset; broker-side __consumer_offsets topic was cleaned.

Related errors

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