Receiving an empty list when using RecordFilterStrategy on batch messages

[m-hetz]

Hi!

I'm using RecordFilterStrategy to filter records based on headers. This works great on single messages, the filter works and records don't reach the listener (annotated with @KafkaListener).

When using batch mode, filtering also works but when the entire batch is filtered, the listener is invoked with an empty list.
Is this expected behavior? According to this when returning an empty list all records are filtered

Is there a way to configure it not to reach the listener at all? (on an empty bulk)

Thanks!

val filter = object : RecordFilterStrategy<String, String> {
    override fun filter(consumerRecord: ConsumerRecord<String, String>): Boolean {
        val header = consumerRecord.headers().lastHeader("SKIP")
        if (header != null) {
            logger.info("Filtering header=" + consumerRecord.key())
            return true
        }
        return false
    }

    override fun filterBatch(records: MutableList<ConsumerRecord<String, String>>): MutableList<ConsumerRecord<String, String>> {
        val iterator: MutableIterator<ConsumerRecord<String, String>> = records.iterator()

        while (iterator.hasNext()) {
            if (this.filter(iterator.next())) {
                iterator.remove()
            }
        }
        if (records.size > 0) return records
        return mutableListOf()
    }
}
factory.setBatchListener(isBatchListener)
factory.setRecordFilterStrategy(filter)


@KafkaListener(topics = ["topic"], containerFactory = "messageContainerFactory")
fun listener(messages: List<ConsumerRecord<String, String?>>,
) {

    logger.info("Consuming messages, batch size=${messages.size}")
}

Sorry for the unformatted code, I tried editing the message but the code is still messed up

[artembilan]

The ConcurrentKafkaListenerContainerFactory has this option for your consideration:

	/**
	 * Set to true to ack discards when a filter strategy is in use.
	 * @param ackDiscarded the ackDiscarded.
	 */
	public void setAckDiscarded(Boolean ackDiscarded) {

The logic then is like this:

		/*
		 *  An empty list goes to the listener if ackDiscarded is false and the listener can ack
		 *  either through the acknowledgment
		 */
		if (consumerRecords.size() > 0 || consumerAware
				|| (!this.ackDiscarded && this.delegateType.equals(ListenerType.ACKNOWLEDGING))) {
			invokeDelegate(consumerRecords, acknowledgment, consumer);
		}
		else {
			if (this.ackDiscarded && acknowledgment != null) {
				acknowledgment.acknowledge();
			}
		}

So, you set that option to true and empty records list is going to be acknowledged automatically and your listener method won't be called.

See more info in docs: https://docs.spring.io/spring-kafka/reference/html/#filtering-messages

[m-hetz]

Thanks I tried that already, I saw that piece of code. When I debug the code I still enter the if because the consumerAware value is true.
This is because the delegateType is of ACKNOWLEDGING_CONSUMER_AWARE since the listener is BatchAcknowledgingConsumerAwareMessageListener. I don't know how to change it or the implications of it

[artembilan]

Ah! I see your point.
An original logic for this kind of listener is like this:

		if (isBatchListener()) {
			BatchMessagingMessageListenerAdapter<K, V> messageListener = new BatchMessagingMessageListenerAdapter<K, V>(
					this.bean, this.method, this.errorHandler);

So, independently of the target method signature we always use a BatchAcknowledgingConsumerAwareMessageListener type.
Therefore the logic in the FilteringBatchMessageListenerAdapter is always true.

Let me create a respective issue and we will discuss with team what can be done on the matter.

Unfortunately there is no way to workaround it for @KafkaListener methods.

Thank you!

[m-hetz]

Ok thank you for that!
I'll subscribe to the issue