Set<String> findConsumerGroups()
            throws InterruptedException, ExecutionException {
        List<String> filteredGroups = listConsumerGroups().stream()
                .map(ConsumerGroupListing::groupId)
                .filter(this::shouldReplicateByGroupFilter)
                .collect(Collectors.toList());

        Set<String> checkpointGroups = new HashSet<>();
        Set<String> irrelevantGroups = new HashSet<>();

        for (String group : filteredGroups) {
            Set<String> consumedTopics = listConsumerGroupOffsets(group).keySet().stream()
                    .map(TopicPartition::topic)
                    .filter(this::shouldReplicateByTopicFilter)
                    .collect(Collectors.toSet());
            // Only perform checkpoints for groups that have offsets for at least one topic that's accepted
            // by the topic filter.
            if (consumedTopics.isEmpty()) {
                irrelevantGroups.add(group);
            } else {
                checkpointGroups.add(group);
            }
        }

        log.debug("Ignoring the following groups which do not have any offsets for topics that are accepted by " +
                        "the topic filter: {}", irrelevantGroups);
        return checkpointGroups;
    }
@Test
    public void testFindConsumerGroups() throws Exception {
        MirrorCheckpointConfig config = new MirrorCheckpointConfig(makeProps());
        MirrorCheckpointConnector connector = new MirrorCheckpointConnector(Collections.emptySet(), config);
        connector = spy(connector);

        Collection<ConsumerGroupListing> groups = Arrays.asList(
                new ConsumerGroupListing("g1", true),
                new ConsumerGroupListing("g2", false));
        Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
        offsets.put(new TopicPartition("t1", 0), new OffsetAndMetadata(0));
        doReturn(groups).when(connector).listConsumerGroups();
        doReturn(true).when(connector).shouldReplicateByTopicFilter(anyString());
        doReturn(true).when(connector).shouldReplicateByGroupFilter(anyString());
        doReturn(offsets).when(connector).listConsumerGroupOffsets(anyString());
        Set<String> groupFound = connector.findConsumerGroups();

        Set<String> expectedGroups = groups.stream().map(ConsumerGroupListing::groupId).collect(Collectors.toSet());
        assertEquals(expectedGroups, groupFound,
                "Expected groups are not the same as findConsumerGroups");

        doReturn(false).when(connector).shouldReplicateByTopicFilter(anyString());
        Set<String> topicFilterGroupFound = connector.findConsumerGroups();
        assertEquals(Collections.emptySet(), topicFilterGroupFound);
    }