apache/kafka · error · IllegalStateException

MockConsumer didn't have duration offset specified, but trie

Error message

MockConsumer didn't have duration offset specified, but tried to seek to timestamp

What it means

Same MockConsumer reset path as the LATEST/BEGINNING cases, but for OffsetResetStrategy.BY_DURATION (timestamp-based reset). MockConsumer consults its durationResetOffsets map; if the test never seeded a timestamp->offset mapping for the partition via updateDurationOffsets(...), it throws this IllegalStateException instead of seeking.

Source

Thrown at clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java:750

            subscriptions.seek(tp, committed.get(tp).offset());
        }
    }

    private void resetOffsetPosition(TopicPartition tp) {
        AutoOffsetResetStrategy strategy = subscriptions.resetStrategy(tp);
        Long offset;
        if (strategy == AutoOffsetResetStrategy.EARLIEST) {
            offset = beginningOffsets.get(tp);
            if (offset == null)
                throw new IllegalStateException("MockConsumer didn't have beginning offset specified, but tried to seek to beginning");
        } else if (strategy == AutoOffsetResetStrategy.LATEST) {
            offset = endOffsets.get(tp);
            if (offset == null)
                throw new IllegalStateException("MockConsumer didn't have end offset specified, but tried to seek to end");
        } else if (strategy.type() == AutoOffsetResetStrategy.StrategyType.BY_DURATION) {
            offset = durationResetOffsets.get(tp);
            if (offset == null)
                throw new IllegalStateException("MockConsumer didn't have duration offset specified, but tried to seek to timestamp");
        } else {
            throw new NoOffsetForPartitionException(tp);
        }
        seek(tp, offset);
    }

    @Override
    public List<PartitionInfo> partitionsFor(String topic, Duration timeout) {
        return partitionsFor(topic);
    }

    @Override
    public Map<String, List<PartitionInfo>> listTopics(Duration timeout) {
        return listTopics();
    }

    @Override
    public Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch,

View on GitHub (pinned to 996fb4585a)

Solutions

1. Call mockConsumer.updateDurationOffsets(Map.of(tp, <offsetAtTimestamp>)) for every partition before the first poll that may trigger a reset.
2. Switch the strategy to EARLIEST or LATEST and use updateBeginningOffsets/updateEndOffsets if duration-based reset is not the behavior under test.
3. Provide a committed offset or call seek(tp, offset) so resetOffsetPosition is bypassed.

Example fix

// before
MockConsumer<byte[],byte[]> c = new MockConsumer<>(OffsetResetStrategy.BY_DURATION);
c.subscribe(Collections.singleton("t"));
c.poll(Duration.ofMillis(0)); // -> IllegalStateException

// after
MockConsumer<byte[],byte[]> c = new MockConsumer<>(OffsetResetStrategy.BY_DURATION);
c.updateDurationOffsets(Map.of(new TopicPartition("t",0), 5L));
c.subscribe(Collections.singleton("t"));
c.poll(Duration.ofMillis(0));

// Ensure duration offsets are seeded for any partition that may reset by duration.
if (mockConsumer.currentStrategy() == OffsetResetStrategy.BY_DURATION) {
    Map<TopicPartition, Long> seed = assigned.stream()
        .collect(Collectors.toMap(tp -> tp, tp -> 0L));
    mockConsumer.updateDurationOffsets(seed);
}

null

null

When it happens

Trigger: Subscription configured with an assignor/strategy whose reset type is BY_DURATION, the partition has no committed offset, and the test never called mockConsumer.updateDurationOffsets(Map.of(tp, offset)) for that partition; poll()/position() then enters resetOffsetPosition(tp) with strategy.type() == BY_DURATION.

Common situations: New tests exercising timestamp-based auto offset reset; copy-pasting an EARLIEST/LATEST test setup and forgetting the duration seed; or asserting offsetsForTimes flow where the reset path is triggered as a side effect.

Related errors

• MockConsumer didn't have end offset specified, but tried to
• clientInstanceId not set
• Cannot add records for a topics that is not subscribed by th
• This consumer has already been closed.
• Invalid negative timestamp

AI-assisted analysis of apache/kafka@996fb4585a (2026-08-11). Data as JSON: /api/errors/282875bdf05542cd. Report an issue: GitHub.