@Override
        public ConsumerRecords<K, V> onConsume(ConsumerRecords<K, V> records) {
            onConsumeCount++;
            if (throwExceptionOnConsume)
                throw new KafkaException("Injected exception in FilterConsumerInterceptor.onConsume.");

            // filters out topic/partitions with partition == FILTER_PARTITION
            Map<TopicPartition, List<ConsumerRecord<K, V>>> recordMap = new HashMap<>();
            for (TopicPartition tp : records.partitions()) {
                if (tp.partition() != filterPartition)
                    recordMap.put(tp, records.records(tp));
            }
            return new ConsumerRecords<>(recordMap);
        }
Test consumer interceptor that filters records in onConsume() intercept
clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerInterceptorsTest.java