apache/kafka · error · UnsupportedOperationException

Not implemented yet.

Error message

Not implemented yet.

What it means

Thrown by MockConsumer.offsetsForTimes unconditionally; this method is not implemented in MockConsumer. Tests that need offset-by-timestamp semantics must either use a real KafkaConsumer in an integration test or supply a custom stub subclass. Calling it on the stock mock always fails.

Source

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

    @Override
    public synchronized void pause(Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions) {
            subscriptions.pause(partition);
            paused.add(partition);
        }
    }

    @Override
    public synchronized void resume(Collection<TopicPartition> partitions) {
        for (TopicPartition partition : partitions) {
            subscriptions.resume(partition);
            paused.remove(partition);
        }
    }

    @Override
    public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> timestampsToSearch) {
        throw new UnsupportedOperationException("Not implemented yet.");
    }

    @Override
    public synchronized Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions) {
        if (offsetsException != null) {
            RuntimeException exception = this.offsetsException;
            this.offsetsException = null;
            throw exception;
        }
        Map<TopicPartition, Long> result = new HashMap<>();
        for (TopicPartition tp : partitions) {
            Long beginningOffset = beginningOffsets.get(tp);
            if (beginningOffset == null)
                throw new IllegalStateException("The partition " + tp + " does not have a beginning offset.");
            result.put(tp, beginningOffset);
        }
        return result;
    }

View on GitHub (pinned to 996fb4585a)

Solutions

1. Move the test to an integration test using a real KafkaConsumer and a broker (or EmbeddedKafka).
2. Subclass MockConsumer and override offsetsForTimes to return a precomputed map.
3. Abstract the offsetsForTimes call behind an interface and stub that interface in the unit test.

Example fix

// before
Map<TopicPartition,OffsetAndTimestamp> r = mock.offsetsForTimes(Map.of(tp, 0L)); // throws

// after (subclass)
MockConsumer<String,String> mock = new MockConsumer<>(OffsetResetStrategy.EARLIEST) {
    @Override
    public synchronized Map<TopicPartition, OffsetAndTimestamp> offsetsForTimes(Map<TopicPartition, Long> tts) {
        return Map.of(tp, new OffsetAndTimestamp(0L, 0L));
    }
};

// No pre-call validation can make offsetsForTimes succeed; detect the path and avoid it.
if (consumer instanceof MockConsumer) {
    throw new UnsupportedOperationException("offsetsForTimes is not supported on MockConsumer; use a subclass or integration test");
}

static boolean supportsOffsetsForTimes(Consumer<?,?> c) {
    return !(c instanceof MockConsumer);
}

try {
    return consumer.offsetsForTimes(timestampsToSearch);
} catch (UnsupportedOperationException e) {
    if ("Not implemented yet.".equals(e.getMessage()) && consumer instanceof MockConsumer) {
        return Collections.emptyMap(); // or move the test to integration
    }
    throw e;
}

When it happens

Trigger: Calling mock.offsetsForTimes(...) in a unit test; exercising consumer code paths that internally invoke offsetsForTimes without a stub.

Common situations: Test code that mirrors production consumer usage verbatim; windowed/time-based consumer logic under test; refactoring that pulls more behavior into a unit test that previously was integration-only.

Related errors

• You can only check the position for partitions assigned to t
• clientInstanceId not set
• The partition {} does not have a beginning offset.
• The partition {} does not have an end offset.
• MockConsumer didn't have beginning offset specified, but tri

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