@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 testFilterOnAbsentTopic() {
        String presentTopic = "present";
        String absentTopic = "absent";
        
        NamedTopicFilter filter = new NamedTopicFilter(presentTopic, absentTopic);
        when(consumerMock.partitionsFor(presentTopic)).thenReturn(Collections.singletonList(createPartitionInfo(presentTopic, 2)));
        when(consumerMock.partitionsFor(absentTopic)).thenReturn(null);
        
        Set<TopicPartition> presentPartitions = filter.getAllSubscribedPartitions(consumerMock);
        assertThat("Expected filter to pass only topics which are present", presentPartitions,
            contains(new TopicPartition(presentTopic, 2)));
    }