public static Map<TopicPartition, Long> fetchCommittedOffsets(final Set<TopicPartition> partitions,
                                                                  final Consumer<byte[], byte[]> consumer) {
        if (partitions.isEmpty()) {
            return Collections.emptyMap();
        }

        final Map<TopicPartition, Long> committedOffsets;
        try {
            // those which do not have a committed offset would default to 0
            committedOffsets = consumer.committed(partitions).entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue() == null ? 0L : e.getValue().offset()));
        } catch (final TimeoutException timeoutException) {
            LOG.warn("The committed offsets request timed out, try increasing the consumer client's default.api.timeout.ms", timeoutException);
            throw timeoutException;
        } catch (final KafkaException fatal) {
            LOG.warn("The committed offsets request failed.", fatal);
            throw new StreamsException(String.format("Failed to retrieve end offsets for %s", partitions), fatal);
        }

        return committedOffsets;
    }
@Test
    public void fetchCommittedOffsetsShouldReturnEmptyMapIfPartitionsAreEmpty() {
        @SuppressWarnings("unchecked")
        final Consumer<byte[], byte[]> consumer = mock(Consumer.class);
        assertTrue(fetchCommittedOffsets(emptySet(), consumer).isEmpty());
    }