apache/kafka · error · IllegalArgumentException

Invalid offset: ${offsetAndMetadata.offset()}

Error message

Invalid offset: ${offsetAndMetadata.offset()}

What it means

Thrown by CommitEvent.validate() when committing offsets and at least one OffsetAndMetadata in the supplied map has a negative offset(). Kafka offsets are non-negative sequence numbers within a partition; a negative value is never a legal commit target and almost always indicates a logic bug in the caller. The check runs in the CommitEvent constructor on the application thread, before the commit is ever enqueued to the network thread.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/internals/events/CommitEvent.java:57

    protected final CompletableFuture<Void> offsetsReady = new CompletableFuture<>();

    protected CommitEvent(final Type type, final Optional<Map<TopicPartition, OffsetAndMetadata>> offsets, final long deadlineMs) {
        super(type, deadlineMs);
        this.offsets = validate(offsets);
    }

    /**
     * Validates the offsets are not negative and then returns the given offset map as
     * {@link Collections#unmodifiableMap(Map) as unmodifiable}.
     */
    private static Optional<Map<TopicPartition, OffsetAndMetadata>> validate(final Optional<Map<TopicPartition, OffsetAndMetadata>> offsets) {
        if (offsets.isEmpty()) {
            return Optional.empty();
        }

        for (OffsetAndMetadata offsetAndMetadata : offsets.get().values()) {
            if (offsetAndMetadata.offset() < 0) {
                throw new IllegalArgumentException("Invalid offset: " + offsetAndMetadata.offset());
            }
        }

        return Optional.of(Collections.unmodifiableMap(offsets.get()));
    }

    public Optional<Map<TopicPartition, OffsetAndMetadata>> offsets() {
        return offsets;
    }

    public CompletableFuture<Void> offsetsReady() {
        return offsetsReady;
    }

    public void markOffsetsReady() {
        offsetsReady.complete(null);
    }

View on GitHub (pinned to c31c9215e1)

Solutions

1. Validate/clamp offsets to >= 0 before constructing OffsetAndMetadata (Math.max(0, computed)).
2. Audit the source of the negative value — it is usually position()-N or end-begin underflow on an empty partition; guard those paths.
3. If you intended 'no offset', skip that partition from the commit map rather than sending a sentinel.
4. Add a unit test asserting every offset you commit is non-negative.

Example fix

// before
long base = consumer.position(tp);
long toCommit = base - 1; // becomes -1 on first poll
consumer.commitSync(Map.of(tp, new OffsetAndMetadata(toCommit)));

// after
long toCommit = Math.max(0, consumer.position(tp));
consumer.commitSync(Map.of(tp, new OffsetAndMetadata(toCommit)));

// Validate offsets are non-negative before handing them to commitSync /
// commitAsync. CommitEvent.validate throws IllegalArgumentException on offset < 0.
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import java.util.Map;

static void ensureValidOffsets(Map<TopicPartition, OffsetAndMetadata> offsets) {
    for (Map.Entry<TopicPartition, OffsetAndMetadata> e : offsets.entrySet()) {
        OffsetAndMetadata om = e.getValue();
        if (om == null || om.offset() < 0) {
            throw new IllegalArgumentException(
                "Refusing to commit invalid offset for " + e.getKey()
                + ": " + (om == null ? "null" : om.offset()));
        }
    }
}

// usage:
ensureValidOffsets(offsets);
consumer.commitSync(offsets);

// Predicate narrowing a raw offset map to a known-safe one (Java has no
// structural type narrowing, so emulate with a validated wrapper).
static boolean allOffsetsValid(Map<TopicPartition, OffsetAndMetadata> m) {
    return m != null && m.values().stream()
        .allMatch(om -> om != null && om.offset() >= 0);
}
// if (allOffsetsValid(offsets)) consumer.commitSync(offsets); else ...

try {
    consumer.commitSync(offsets);
} catch (IllegalArgumentException e) {
    // offset < 0 slipped through; drop or reset the offending partition
    log.warn("Rejected bad offset on commit; skipping: {}", e.getMessage());
}

When it happens

Trigger: consumer.commitSync(Map<TopicPartition,OffsetAndMetadata>) or commitAsync(...) with a hand-built map containing a negative offset; building OffsetAndMetadata from an arithmetic expression (e.g. position()-1) that underflows; using -1 as an 'unset' sentinel and passing it through to commit.

Common situations: Custom offset-tracking logic that subtracts from position()/offset and goes below zero on the first message; porting code that used -1 as a null marker; replay-from-offset tooling that loads offsets from a corrupt store; off-by-one in 'last committed + delta' calculations when the partition is empty.

Related errors

• Invalid negative offset
• The target time for partition {} is {}. The target time cann
• Invalid negative offset
• Invalid negative timestamp
• Invalid value `{}` for configuration {}. The value must be e

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