@Override
    public Set<TopicPartition> getAllSubscribedPartitions(Consumer<?, ?> consumer) {
        Set<TopicPartition> allPartitions = new HashSet<>();
        for (String topic : topics) {
            List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic);
            if (partitionInfoList != null) {
                for (PartitionInfo partitionInfo : partitionInfoList) {
                    allPartitions.add(new TopicPartition(partitionInfo.topic(), partitionInfo.partition()));
                }
            } else {
                LOG.warn("Topic {} not found, skipping addition of the topic", topic);
            }
        }
        return allPartitions;
    }
@Test
    public void testFilter() {
        String matchingTopicOne = "test-1";
        String matchingTopicTwo = "test-11";
        String unmatchedTopic = "unmatched";
        
        NamedTopicFilter filter = new NamedTopicFilter(matchingTopicOne, matchingTopicTwo);
        
        when(consumerMock.partitionsFor(matchingTopicOne)).thenReturn(Collections.singletonList(createPartitionInfo(matchingTopicOne, 0)));
        List<PartitionInfo> partitionTwoPartitions = new ArrayList<>();
        partitionTwoPartitions.add(createPartitionInfo(matchingTopicTwo, 0));
        partitionTwoPartitions.add(createPartitionInfo(matchingTopicTwo, 1));
        when(consumerMock.partitionsFor(matchingTopicTwo)).thenReturn(partitionTwoPartitions);
        when(consumerMock.partitionsFor(unmatchedTopic)).thenReturn(Collections.singletonList(createPartitionInfo(unmatchedTopic, 0)));
        
        Set<TopicPartition> matchedPartitions = filter.getAllSubscribedPartitions(consumerMock);
        
        assertThat("Expected filter to pass only topics with exact name matches", matchedPartitions, 
            containsInAnyOrder(new TopicPartition(matchingTopicOne, 0), new TopicPartition(matchingTopicTwo, 0), new TopicPartition(matchingTopicTwo, 1)));
            
    }