Receiving an empty list when using RecordFilterStrategy on batch messages

[artembilan]

Discussed in #2805

^{Originally posted by m-hetz September 11, 2023}
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

[chickenchickenlove]

Hi, @artembilan.
I'm interested in solving this issue.

		if (consumerRecords.size() > 0 || consumerAware
				|| (!this.ackDiscarded && this.delegateType.equals(ListenerType.ACKNOWLEDGING))) {
			invokeDelegate(consumerRecords, acknowledgment, consumer);
		}

Why we should consider consumerAware type even if consumerRecords.size() == 0?

		if (!filter(consumerRecord)) {
			switch (this.delegateType) {
				case ACKNOWLEDGING_CONSUMER_AWARE -> this.delegate.onMessage(consumerRecord, acknowledgment, consumer);
				case ACKNOWLEDGING -> this.delegate.onMessage(consumerRecord, acknowledgment);
				case CONSUMER_AWARE -> this.delegate.onMessage(consumerRecord, consumer);
				case SIMPLE -> this.delegate.onMessage(consumerRecord);
			}
		}
		else {
			ackFilteredIfNecessary(acknowledgment);
		}

In a not batch mode, if record are filtered all, MessageListenerAdapter does not care about at all.
Thus what do you think to fix code like this? (please note if condition)

                 // Fix Here. consumerRecords.size() should be grater than 0. 
		if ((consumerRecords.size() > 0) && (consumerAware
				|| (!this.ackDiscarded && this.delegateType.equals(ListenerType.ACKNOWLEDGING)))) {
			invokeDelegate(consumerRecords, acknowledgment, consumer);
		}
                // And then, consumerRecords.size() == 0, ackFilteredIfNecessary! 
		else {
			if (this.ackDiscarded && acknowledgment != null) {
				acknowledgment.acknowledge();
			}
		}

IMHO, it is not necessary to be aware of consumer if filtered record size is 0. This is because it is already being ignored in single mode.
What do you think?

[sobychacko]

@chickenchickenlove You can go ahead and work on a PR. Then, we can review the details. Keep in mind that we are already in an RC state with 3.2.0. Therefore, any new enhancements like these must wait for the 3.3.0 version.