@Override
    public Map<TopicPartition, Long> beginningOffsets(Collection<TopicPartition> partitions) {
        return beginningOffsets(partitions, Duration.ofMillis(defaultApiTimeoutMs));
    }
@Test
    public void testBeginningOffsets() {
        consumer = newConsumer();
        Map<TopicPartition, OffsetAndTimestampInternal> expectedOffsets = mockOffsetAndTimestamp();

        when(applicationEventHandler.addAndGet(any(ListOffsetsEvent.class))).thenAnswer(invocation -> {
            ListOffsetsEvent event = invocation.getArgument(0);
            Timer timer = time.timer(event.deadlineMs() - time.milliseconds());
            if (timer.remainingMs() == 0) {
                fail("Timer duration should not be zero.");
            }
            return expectedOffsets;
        });

        Map<TopicPartition, Long> result = assertDoesNotThrow(() -> consumer.beginningOffsets(expectedOffsets.keySet(), Duration.ofMillis(1)));

        expectedOffsets.forEach((key, value) -> {
            assertTrue(result.containsKey(key));
            assertEquals(value.offset(), result.get(key));
        });
        verify(applicationEventHandler).addAndGet(any(ListOffsetsEvent.class));
    }