public static Map<TopicPartition, ListOffsetsResultInfo> fetchEndOffsets(final Collection<TopicPartition> partitions,
                                                                             final Admin adminClient) {
        if (partitions.isEmpty()) {
            return Collections.emptyMap();
        }
        return getEndOffsets(fetchEndOffsetsFuture(partitions, adminClient));
    }
@Test
    public void fetchEndOffsetsShouldRethrowExecutionExceptionAsStreamsException() throws Exception {
        final Admin adminClient = mock(AdminClient.class);
        final ListOffsetsResult result = mock(ListOffsetsResult.class);
        @SuppressWarnings("unchecked")
        final KafkaFuture<Map<TopicPartition, ListOffsetsResultInfo>> allFuture = mock(KafkaFuture.class);

        when(adminClient.listOffsets(any())).thenReturn(result);
        when(result.all()).thenReturn(allFuture);
        when(allFuture.get()).thenThrow(new ExecutionException(new RuntimeException()));

        assertThrows(StreamsException.class, () -> fetchEndOffsets(PARTITIONS, adminClient));
    }