@Test
public void testBasicConsumerWithRecordFilter() {

    final int maxRecordsPerPoll = 10;
    // Create a topic with 2 partitions, (partitionId 0, 1)
    final String topicName = "TestTopic";
    sharedKafkaTestResource
        .getKafkaTestUtils()
        .createTopic(topicName, 2, (short) 1);
    // Produce 10 records into partition 0 of topic.
    sharedKafkaTestResource
        .getKafkaTestUtils()
        .produceRecords(maxRecordsPerPoll, topicName, 0);
    // Produce 10 records into partition 1 of topic.
    sharedKafkaTestResource
        .getKafkaTestUtils()
        .produceRecords(maxRecordsPerPoll, topicName, 1);
    // Create cluster Config
    final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
        .withBrokerHosts(sharedKafkaTestResource.getKafkaConnectString())
        .build();
    // Create Deserializer Config
    final DeserializerConfig deserializerConfig = DeserializerConfig.newBuilder()
        .withKeyDeserializerClass(StringDeserializer.class)
        .withValueDeserializerClass(StringDeserializer.class)
        .build();
    // Create Topic Config
    final TopicConfig topicConfig = new TopicConfig(clusterConfig, deserializerConfig, topicName);
    // Create FilterConfig
    final FilterConfig filterConfig = FilterConfig.withFilters(
        new RecordFilterDefinition(new PartitionFilter(), new HashMap<>())
    );
    // Create ClientConfig
    final ClientConfig clientConfig = ClientConfig.newBuilder()
        .withConsumerId("MyConsumerId")
        .withFilterConfig(filterConfig)
        .withAllPartitions()
        .withMaxResultsPerPartition(maxRecordsPerPoll * 100)
        .withTopicConfig(topicConfig)
        .build();
    // Create consumer
    try (final KafkaConsumer<String, String> consumer = kafkaConsumerFactory.createConsumerAndSubscribe(clientConfig)) {
        // Attempt to consume, should pull first 10
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));
        assertEquals("Should have found " + maxRecordsPerPoll + " records", maxRecordsPerPoll, records.count());
        for (final ConsumerRecord<String, String> record: records) {
            assertEquals("Should be from parittion 1 only", 1, record.partition());
        }
        // Attempt to consume, should come up empty
        records = consumer.poll(Duration.ofSeconds(2));
        assertTrue("Should be empty", records.isEmpty());
    }
}