{"record":{"id":"d34130bd9a741391","repo":"apache/kafka","slug":"mockconsumer-didn-t-have-end-offset-specified-but","errorCode":null,"errorMessage":"MockConsumer didn't have end offset specified, but tried to seek to end","messagePattern":"MockConsumer didn't have end offset specified, but tried to seek to end","errorType":"exception","errorClass":"IllegalStateException","httpStatus":null,"severity":"error","filePath":"clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java","lineNumber":746,"sourceCode":"        } else if (!committed.containsKey(tp)) {\n            subscriptions.requestOffsetReset(tp);\n            resetOffsetPosition(tp);\n        } else {\n            subscriptions.seek(tp, committed.get(tp).offset());\n        }\n    }\n\n    private void resetOffsetPosition(TopicPartition tp) {\n        AutoOffsetResetStrategy strategy = subscriptions.resetStrategy(tp);\n        Long offset;\n        if (strategy == AutoOffsetResetStrategy.EARLIEST) {\n            offset = beginningOffsets.get(tp);\n            if (offset == null)\n                throw new IllegalStateException(\"MockConsumer didn't have beginning offset specified, but tried to seek to beginning\");\n        } else if (strategy == AutoOffsetResetStrategy.LATEST) {\n            offset = endOffsets.get(tp);\n            if (offset == null)\n                throw new IllegalStateException(\"MockConsumer didn't have end offset specified, but tried to seek to end\");\n        } else if (strategy.type() == AutoOffsetResetStrategy.StrategyType.BY_DURATION) {\n            offset = durationResetOffsets.get(tp);\n            if (offset == null)\n                throw new IllegalStateException(\"MockConsumer didn't have duration offset specified, but tried to seek to timestamp\");\n        } else {\n            throw new NoOffsetForPartitionException(tp);\n        }\n        seek(tp, offset);\n    }\n\n    @Override\n    public List<PartitionInfo> partitionsFor(String topic, Duration timeout) {\n        return partitionsFor(topic);\n    }\n\n    @Override\n    public Map<String, List<PartitionInfo>> listTopics(Duration timeout) {\n        return listTopics();","sourceCodeStart":728,"sourceCodeEnd":764,"githubUrl":"https://github.com/apache/kafka/blob/996fb4585aa1bcc8980b0e1b8d6b168b986cd979/clients/src/main/java/org/apache/kafka/clients/consumer/MockConsumer.java#L728-L764","documentation":"MockConsumer is a test double for KafkaConsumer that simulates offset resets without a broker. When the subscription's AutoOffsetResetStrategy is LATEST, MockConsumer looks up the partition's end offset from its internal endOffsets map (seeded by the test). If no end offset has been recorded for that TopicPartition, it throws this IllegalStateException because it cannot decide where to seek.","triggerScenarios":"Calling poll()/position() on a MockConsumer whose subscription uses OffsetResetStrategy.LATEST (the default for new MockConsumer(OffsetResetStrategy.LATEST)), for a TopicPartition that was never seeded via updateEndOffsets(...), and for which no committed offset exists, so the consumer falls through to resetOffsetPosition(tp) and strategy == LATEST.","commonSituations":"Writing KafkaConsumer unit tests and forgetting to call updateEndOffsets(Map.of(tp, 0L)) before the first poll; switching the test from EARLIEST to LATEST without re-seeding offsets; or assigning a new partition mid-test (rebalance) whose end offset was never provided.","solutions":["Before poll(), call mockConsumer.updateEndOffsets(Map.of(tp, <endOffsetLong>)) for every partition the consumer will be assigned.","If you only need earliest behavior, construct the MockConsumer with OffsetResetStrategy.EARLIEST and call updateBeginningOffsets(...) instead.","If you do not want a reset to happen at all, seed a committed offset via commitSync(...) or call seek(tp, offset) explicitly before poll so resetOffsetPosition is never reached.","After any rebalance/subscription change in the test, re-seed end offsets for the newly assigned partitions."],"exampleFix":"// before\nMockConsumer<byte[],byte[]> c = new MockConsumer<>(OffsetResetStrategy.LATEST);\nc.subscribe(Collections.singleton(\"t\"));\nc.poll(Duration.ofMillis(0)); // -> IllegalStateException\n\n// after\nMockConsumer<byte[],byte[]> c = new MockConsumer<>(OffsetResetStrategy.LATEST);\nc.updateEndOffsets(Map.of(new TopicPartition(\"t\",0), 10L));\nc.subscribe(Collections.singleton(\"t\"));\nc.poll(Duration.ofMillis(0));","handlingStrategy":"validation","validationCode":"// Before poll(), ensure every assigned partition has an end offset seeded.\nSet<TopicPartition> assigned = mockConsumer.assignment();\nMap<TopicPartition, Long> missing = assigned.stream()\n    .filter(tp -> !endOffsetsSeeded.contains(tp))\n    .collect(Collectors.toMap(tp -> tp, tp -> 0L));\nif (!missing.isEmpty()) {\n    mockConsumer.updateEndOffsets(missing);\n}","typeGuard":"null","tryCatchPattern":"null","preventionTips":["Always call updateEndOffsets(...) for every partition right after subscribe/assign in MockConsumer tests.","Centralize MockConsumer setup in a helper that seeds both beginning and end offsets so a strategy change does not break the test.","After a manual rebalance() in the test, re-seed offsets for the new assignment."],"tags":["testing","mock","consumer","offset-reset","kafka-clients"],"backgroundTag":null,"analyzedSha":"996fb4585aa1bcc8980b0e1b8d6b168b986cd979","analyzedAt":"2026-08-11T22:03:28.655Z","contentChangedAt":null,"schemaVersion":2},"datasetVersion":"2026-09-14T00:17:10.932Z"}