public void awaitFutures(Collection<TopicPartition> topicPartitions) {
        futuresFor(topicPartitions).forEach(future -> {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                log.error("Encountered an error while awaiting an errant record future's completion.", e);
                throw new ConnectException(e);
            }
        });
    }
@Test
    public void testGetFutures() {
        initializeReporter(true);
        Collection<TopicPartition> topicPartitions = new ArrayList<>();
        assertTrue(reporter.futures.isEmpty());
        for (int i = 0; i < 4; i++) {
            TopicPartition topicPartition = new TopicPartition("topic", i);
            topicPartitions.add(topicPartition);
            reporter.futures.put(topicPartition, Collections.singletonList(CompletableFuture.completedFuture(null)));
        }
        assertFalse(reporter.futures.isEmpty());
        reporter.awaitFutures(topicPartitions);
        assertTrue(reporter.futures.isEmpty());
    }