@Override
    public Set<TopicPartition> getAllSubscribedPartitions(Consumer<?, ?> consumer) {
        topics.clear();
        Set<TopicPartition> allPartitions = new HashSet<>();
        for (Map.Entry<String, List<PartitionInfo>> entry : consumer.listTopics().entrySet()) {
            if (pattern.matcher(entry.getKey()).matches()) {
                for (PartitionInfo partitionInfo : entry.getValue()) {
                    allPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
                    topics.add(partitionInfo.topic());
                }
            }
        }
        return allPartitions;
    }
@Test
    public void testFilter() {
        Pattern pattern = Pattern.compile("test-\\d+");
        PatternTopicFilter filter = new PatternTopicFilter(pattern);
        
        String matchingTopicOne = "test-1";
        String matchingTopicTwo = "test-11";
        String unmatchedTopic = "unmatched";
        
        Map<String, List<PartitionInfo>> allTopics = new HashMap<>();
        allTopics.put(matchingTopicOne, Collections.singletonList(createPartitionInfo(matchingTopicOne, 0)));
        List<PartitionInfo> testTwoPartitions = new ArrayList<>();
        testTwoPartitions.add(createPartitionInfo(matchingTopicTwo, 0));
        testTwoPartitions.add(createPartitionInfo(matchingTopicTwo, 1));
        allTopics.put(matchingTopicTwo, testTwoPartitions);
        allTopics.put(unmatchedTopic, Collections.singletonList(createPartitionInfo(unmatchedTopic, 0)));
        
        when(consumerMock.listTopics()).thenReturn(allTopics);
        
        Set<TopicPartition> matchedPartitions = filter.getAllSubscribedPartitions(consumerMock);
        
        assertThat("Expected topic partitions matching the pattern to be passed by the filter", matchedPartitions,
            containsInAnyOrder(new TopicPartition(matchingTopicOne, 0), new TopicPartition(matchingTopicTwo, 0), new TopicPartition(matchingTopicTwo, 1)));
    }