diff --git a/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/filtering.adoc b/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/filtering.adoc index dd65d477..d689bdc5 100644 --- a/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/filtering.adoc +++ b/spring-kafka-docs/src/main/antora/modules/ROOT/pages/kafka/receiving-messages/filtering.adoc @@ -27,3 +27,63 @@ public void listen(Thing thing) { } ---- +Starting with version 3.3, Ignoring empty batches that result from filtering by `RecordFilterStrategy` is supported. +When implementing `RecordFilterStrategy`, it can be configured through `ignoreEmptyBatch()`. +The default setting is `false`, indicating `KafkaListener` will be invoked even if all `ConsumerRecord` s are filtered out. + +If `true` is returned, the `KafkaListener` [underline]#will not be invoked# when all `ConsumerRecord` are filtered out. +However, commit to broker, will still be executed. + +If `false` is returned, the `KafkaListener` [underline]#will be invoked# when all `ConsumerRecord` are filtered out. + +Here are some examples. + +[source,java] +---- +public class IgnoreEmptyBatchRecordFilterStrategy implements RecordFilterStrategy { + ... + @Override + public List> filterBatch( + List> consumerRecords) { + return List.of(); + } + + @Override + public boolean ignoreEmptyBatch() { + return true; + } +}; + +// NOTE: ignoreEmptyBatchRecordFilterStrategy is bean name of IgnoreEmptyBatchRecordFilterStrategy instance. +@KafkaListener(id = "filtered", topics = "topic", filter = "ignoreEmptyBatchRecordFilterStrategy") +public void listen(List things) { + ... +} +---- +In this case, `IgnoreEmptyBatchRecordFilterStrategy` always returns empty list and return `true` as result of `ignoreEmptyBatch()`. +Thus `KafkaListener#listen(...)` never will be invoked at all. + +[source,java] +---- +public class NotIgnoreEmptyBatchRecordFilterStrategy implements RecordFilterStrategy { + ... + @Override + public List> filterBatch( + List> consumerRecords) { + return List.of(); + } + + @Override + public boolean ignoreEmptyBatch() { + return false; + } +}; + +// NOTE: notIgnoreEmptyBatchRecordFilterStrategy is bean name of NotIgnoreEmptyBatchRecordFilterStrategy instance. +@KafkaListener(id = "filtered", topics = "topic", filter = "notIgnoreEmptyBatchRecordFilterStrategy") +public void listen(List things) { + ... +} +---- +However, in this case, `IgnoreEmptyBatchRecordFilterStrategy` always returns empty list and return `false` as result of `ignoreEmptyBatch()`. +Thus `KafkaListener#listen(...)` always will be invoked. diff --git a/spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc b/spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc index 345f1e99..5688b7b4 100644 --- a/spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc +++ b/spring-kafka-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc @@ -19,4 +19,9 @@ A new method, `getGroupId()`, has been added to the `ConsumerSeekCallback` inter This method allows for more selective seek operations by targeting only the desired consumer group. For more details, see xref:kafka/seek.adoc#seek[Seek API Docs]. +[[x33-new-option-ignore-empty-batch]] +=== Configurable Handling of Empty Batches in Kafka Listener with RecordFilterStrategy +`RecordFilterStrategy` now supports ignoring empty batches that result from filtering. +This can be configured through overriding default method `ignoreEmptyBatch()`, which defaults to false, ensuring `KafkaListener` is invoked even if all `ConsumerRecords` are filtered out. +For more details, see xref:kafka/receiving-messages/filtering.adoc[Message receive filtering Docs]. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java index a7eb2c03..e3f09708 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/FilteringBatchMessageListenerAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -36,6 +36,7 @@ import org.springframework.util.Assert; * @param the value type. * * @author Gary Russell + * @author Sanghyeok An * */ public class FilteringBatchMessageListenerAdapter @@ -44,6 +45,8 @@ public class FilteringBatchMessageListenerAdapter private final boolean ackDiscarded; + private final boolean consumerAware; + /** * Create an instance with the supplied strategy and delegate listener. * @param delegate the delegate. @@ -51,9 +54,7 @@ public class FilteringBatchMessageListenerAdapter */ public FilteringBatchMessageListenerAdapter(BatchMessageListener delegate, RecordFilterStrategy recordFilterStrategy) { - - super(delegate, recordFilterStrategy); - this.ackDiscarded = false; + this(delegate, recordFilterStrategy, false); } /** @@ -71,22 +72,25 @@ public class FilteringBatchMessageListenerAdapter super(delegate, recordFilterStrategy); this.ackDiscarded = ackDiscarded; + this.consumerAware = this.delegateType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE) || + this.delegateType.equals(ListenerType.CONSUMER_AWARE); } @Override public void onMessage(List> records, @Nullable Acknowledgment acknowledgment, Consumer consumer) { - List> consumerRecords = getRecordFilterStrategy().filterBatch(records); + final RecordFilterStrategy recordFilterStrategy = getRecordFilterStrategy(); + final List> consumerRecords = recordFilterStrategy.filterBatch(records); Assert.state(consumerRecords != null, "filter returned null from filterBatch"); - boolean consumerAware = this.delegateType.equals(ListenerType.ACKNOWLEDGING_CONSUMER_AWARE) - || this.delegateType.equals(ListenerType.CONSUMER_AWARE); - /* - * 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))) { + + if (recordFilterStrategy.ignoreEmptyBatch() && + consumerRecords.isEmpty() && + acknowledgment != null) { + acknowledgment.acknowledge(); + } + else if (!consumerRecords.isEmpty() || this.consumerAware + || (!this.ackDiscarded && this.delegateType.equals(ListenerType.ACKNOWLEDGING))) { invokeDelegate(consumerRecords, acknowledgment, consumer); } else { @@ -98,6 +102,7 @@ public class FilteringBatchMessageListenerAdapter private void invokeDelegate(List> consumerRecords, Acknowledgment acknowledgment, Consumer consumer) { + switch (this.delegateType) { case ACKNOWLEDGING_CONSUMER_AWARE: this.delegate.onMessage(consumerRecords, acknowledgment, consumer); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java index f2de862e..db7d29dc 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/adapter/RecordFilterStrategy.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,11 +16,12 @@ package org.springframework.kafka.listener.adapter; -import java.util.Iterator; import java.util.List; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.springframework.kafka.listener.BatchMessageListener; + /** * Implementations of this interface can signal that a record about * to be delivered to a message listener should be discarded instead @@ -30,7 +31,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; * @param the value type. * * @author Gary Russell - * + * @author Sanghyeok An */ public interface RecordFilterStrategy { @@ -49,13 +50,20 @@ public interface RecordFilterStrategy { * @since 2.8 */ default List> filterBatch(List> records) { - Iterator> iterator = records.iterator(); - while (iterator.hasNext()) { - if (filter(iterator.next())) { - iterator.remove(); - } - } + records.removeIf(this::filter); return records; } + /** + * Determine whether {@link FilteringBatchMessageListenerAdapter} should invoke + * the {@link BatchMessageListener} when all {@link ConsumerRecord}s in a batch have been filtered out + * resulting in empty list. By default, do invoke the {@link BatchMessageListener} (return false). + * @return true for {@link FilteringBatchMessageListenerAdapter} to {@link BatchMessageListener} + * when all {@link ConsumerRecord} in a batch filtered out + * @since 3.3 + */ + default boolean ignoreEmptyBatch() { + return false; + } + } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/FilteringAdapterTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/FilteringAdapterTests.java index 2b04bc3b..17cb070b 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/FilteringAdapterTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/adapter/FilteringAdapterTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2019 the original author or authors. + * Copyright 2017-2024 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -21,6 +21,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.BDDMockito.willAnswer; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.only; import static org.mockito.Mockito.verify; import java.util.ArrayList; @@ -29,6 +30,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.KafkaConsumer; import org.junit.jupiter.api.Test; import org.springframework.kafka.listener.BatchAcknowledgingMessageListener; @@ -36,6 +38,7 @@ import org.springframework.kafka.support.Acknowledgment; /** * @author Gary Russell + * @author Sanghyeok An * @since 2.0 * */ @@ -46,7 +49,7 @@ public class FilteringAdapterTests { public void testBatchFilter() throws Exception { BatchAcknowledgingMessageListener listener = mock(BatchAcknowledgingMessageListener.class); FilteringBatchMessageListenerAdapter adapter = - new FilteringBatchMessageListenerAdapter(listener, r -> false); + new FilteringBatchMessageListenerAdapter<>(listener, r -> false); List> consumerRecords = new ArrayList<>(); final CountDownLatch latch = new CountDownLatch(1); willAnswer(i -> { @@ -64,12 +67,174 @@ public class FilteringAdapterTests { public void testBatchFilterAckDiscard() throws Exception { BatchAcknowledgingMessageListener listener = mock(BatchAcknowledgingMessageListener.class); FilteringBatchMessageListenerAdapter adapter = - new FilteringBatchMessageListenerAdapter(listener, r -> false, true); + new FilteringBatchMessageListenerAdapter<>(listener, r -> false, true); List> consumerRecords = new ArrayList<>(); final CountDownLatch latch = new CountDownLatch(1); - adapter.onMessage(consumerRecords, () -> latch.countDown(), null); + adapter.onMessage(consumerRecords, latch::countDown, null); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); verify(listener, never()).onMessage(any(List.class), any(Acknowledgment.class)); } + @SuppressWarnings("unchecked") + @Test + public void listener_should_not_be_invoked_on_emptyList_and_ignoreEmptyBatch_true() { + // Given : + RecordFilterStrategy filter = new RecordFilterStrategy<>() { + + @Override + public boolean filter(ConsumerRecord consumerRecord) { + return true; + } + + @Override + public List> filterBatch( + List> consumerRecords) { + return List.of(); + } + + @Override + public boolean ignoreEmptyBatch() { + return true; + } + }; + + BatchAcknowledgingMessageListener listener = mock(); + FilteringBatchMessageListenerAdapter adapter = + new FilteringBatchMessageListenerAdapter<>(listener, filter); + List> consumerRecords = new ArrayList<>(); + Acknowledgment ack = mock(); + + // When : + adapter.onMessage(consumerRecords, ack, null); + + // Then + verify(ack, only()).acknowledge(); + verify(listener, never()).onMessage(any(List.class), any(Acknowledgment.class), any(KafkaConsumer.class)); + verify(listener, never()).onMessage(any(List.class), any(Acknowledgment.class)); + verify(listener, never()).onMessage(any(List.class), any(KafkaConsumer.class)); + verify(listener, never()).onMessage(any(List.class)); + } + + @SuppressWarnings("unchecked") + @Test + public void listener_should_be_invoked_on_notEmptyList_and_ignoreEmptyBatch_true() throws Exception { + // Given : + RecordFilterStrategy filter = new RecordFilterStrategy<>() { + + @Override + public boolean filter(ConsumerRecord consumerRecord) { + return true; + } + + @Override + public List> filterBatch( + List> consumerRecords) { + return consumerRecords; + } + + @Override + public boolean ignoreEmptyBatch() { + return true; + } + }; + + BatchAcknowledgingMessageListener listener = mock(); + FilteringBatchMessageListenerAdapter adapter = + new FilteringBatchMessageListenerAdapter<>(listener, filter); + List> consumerRecords = + List.of(new ConsumerRecord<>("hello-topic", 1, 1, "hello-key", "hello-value")); + Acknowledgment ack = mock(); + + CountDownLatch latch = new CountDownLatch(1); + willAnswer(i -> { + latch.countDown(); + return null; + }).given(listener).onMessage(any(List.class), any(Acknowledgment.class)); + + // When : + adapter.onMessage(consumerRecords, ack, null); + + // Then + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + verify(ack, never()).acknowledge(); + } + + @SuppressWarnings("unchecked") + @Test + public void listener_should_be_invoked_on_emptyList_and_ignoreEmptyBatch_false() throws Exception { + // Given : + RecordFilterStrategy filter = new RecordFilterStrategy<>() { + + @Override + public boolean filter(ConsumerRecord consumerRecord) { + return true; + } + + @Override + public List> filterBatch( + List> consumerRecords) { + return List.of(); + } + }; + + BatchAcknowledgingMessageListener listener = mock(); + FilteringBatchMessageListenerAdapter adapter = + new FilteringBatchMessageListenerAdapter<>(listener, filter); + List> consumerRecords = new ArrayList<>(); + Acknowledgment ack = mock(); + + CountDownLatch latch = new CountDownLatch(1); + willAnswer(i -> { + latch.countDown(); + return null; + }).given(listener).onMessage(any(List.class), any(Acknowledgment.class)); + + // When : + adapter.onMessage(consumerRecords, ack, null); + + // Then + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + verify(ack, never()).acknowledge(); + } + + @SuppressWarnings("unchecked") + @Test + public void listener_should_be_invoked_on_notEmptyList_and_ignoreEmptyBatch_false() throws Exception { + // Given : + RecordFilterStrategy filter = new RecordFilterStrategy<>() { + + @Override + public boolean filter(ConsumerRecord consumerRecord) { + return true; + } + + @Override + public List> filterBatch( + // System Under Test + List> consumerRecords) { + return consumerRecords; + } + }; + + BatchAcknowledgingMessageListener listener = mock(); + FilteringBatchMessageListenerAdapter adapter = + new FilteringBatchMessageListenerAdapter<>(listener, filter); + List> consumerRecords = + List.of(new ConsumerRecord<>("hello-topic", 1, 1, "hello-key", "hello-value")); + Acknowledgment ack = mock(); + + final CountDownLatch latch = new CountDownLatch(1); + willAnswer(i -> { + latch.countDown(); + return null; + }).given(listener).onMessage(any(List.class), any(Acknowledgment.class)); + + // When : + adapter.onMessage(consumerRecords, ack, null); + + // Then + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + verify(ack, never()).acknowledge(); + } + }