diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java index 4deb3c65..9004ebd6 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java @@ -36,6 +36,7 @@ import org.springframework.kafka.listener.BatchErrorHandler; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.ErrorHandler; import org.springframework.kafka.listener.GenericErrorHandler; +import org.springframework.kafka.listener.RecordInterceptor; import org.springframework.kafka.listener.adapter.RecordFilterStrategy; import org.springframework.kafka.listener.adapter.ReplyHeadersConfigurer; import org.springframework.kafka.requestreply.ReplyingKafkaOperations; @@ -95,6 +96,8 @@ public abstract class AbstractKafkaListenerContainerFactory recordInterceptor; + /** * Specify a {@link ConsumerFactory} to use. * @param consumerFactory The consumer factory. @@ -266,6 +269,16 @@ public abstract class AbstractKafkaListenerContainerFactory recordInterceptor) { + this.recordInterceptor = recordInterceptor; + } + @Override public void afterPropertiesSet() { if (this.errorHandler != null) { @@ -363,6 +376,7 @@ public abstract class AbstractKafkaListenerContainerFactory private AfterRollbackProcessor afterRollbackProcessor = new DefaultAfterRollbackProcessor<>(); + private RecordInterceptor recordInterceptor; + private volatile boolean running = false; private volatile boolean paused; @@ -261,6 +263,20 @@ public abstract class AbstractMessageListenerContainer return this.beanName; // the container factory sets the bean name to the id attribute } + protected RecordInterceptor getRecordInterceptor() { + return this.recordInterceptor; + } + + /** + * Set an interceptor to be called before calling the listener. + * Does not apply to batch listeners. + * @param recordInterceptor the interceptor. + * @since 2.2.7 + */ + public void setRecordInterceptor(RecordInterceptor recordInterceptor) { + this.recordInterceptor = recordInterceptor; + } + @Override public void setupMessageListener(Object messageListener) { this.containerProperties.setMessageListener(messageListener); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index 5eb94394..b9061bcc 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -161,6 +161,7 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis container.setClientIdSuffix("-" + i); container.setGenericErrorHandler(getGenericErrorHandler()); container.setAfterRollbackProcessor(getAfterRollbackProcessor()); + container.setRecordInterceptor(getRecordInterceptor()); container.setEmergencyStop(() -> { stop(() -> { // NOSONAR diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index 70aeeae7..9b7c5174 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -465,6 +465,8 @@ public class KafkaMessageListenerContainer // NOSONAR comment density private final boolean checkNullValueForExceptions; + private final RecordInterceptor recordInterceptor = getRecordInterceptor(); + private Map definedPartitions; private volatile Collection assignedPartitions; @@ -1257,26 +1259,37 @@ public class KafkaMessageListenerContainer // NOSONAR comment density ackCurrent(record, producer); } - private void doInvokeOnMessage(final ConsumerRecord record) { - switch (this.listenerType) { - case ACKNOWLEDGING_CONSUMER_AWARE: - this.listener.onMessage(record, - this.isAnyManualAck - ? new ConsumerAcknowledgment(record) - : null, this.consumer); - break; - case CONSUMER_AWARE: - this.listener.onMessage(record, this.consumer); - break; - case ACKNOWLEDGING: - this.listener.onMessage(record, - this.isAnyManualAck - ? new ConsumerAcknowledgment(record) - : null); - break; - case SIMPLE: - this.listener.onMessage(record); - break; + private void doInvokeOnMessage(final ConsumerRecord recordArg) { + ConsumerRecord record = recordArg; + if (this.recordInterceptor != null) { + record = this.recordInterceptor.intercept(record); + } + if (record == null) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("RecordInterceptor returned null, skipping: " + recordArg); + } + } + else { + switch (this.listenerType) { + case ACKNOWLEDGING_CONSUMER_AWARE: + this.listener.onMessage(record, + this.isAnyManualAck + ? new ConsumerAcknowledgment(record) + : null, this.consumer); + break; + case CONSUMER_AWARE: + this.listener.onMessage(record, this.consumer); + break; + case ACKNOWLEDGING: + this.listener.onMessage(record, + this.isAnyManualAck + ? new ConsumerAcknowledgment(record) + : null); + break; + case SIMPLE: + this.listener.onMessage(record); + break; + } } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/RecordInterceptor.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/RecordInterceptor.java new file mode 100644 index 00000000..cc06f886 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/RecordInterceptor.java @@ -0,0 +1,46 @@ +/* + * Copyright 2019 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.listener; + +import org.apache.kafka.clients.consumer.ConsumerRecord; + +import org.springframework.lang.Nullable; + +/** + * An interceptor for {@link ConsumerRecord} invoked by the listener + * container before invoking the listener. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * @since 2.2.7 + * + */ +@FunctionalInterface +public interface RecordInterceptor { + + /** + * Perform some action on the record or return a different one. + * If null is returned the record will be skipped. + * @param record the record. + * @return the record or null. + */ + @Nullable + ConsumerRecord intercept(ConsumerRecord record); + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 1bf8ba27..b692ab92 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -740,6 +740,7 @@ public class EnableKafkaIntegrationTests { this.bytesKeyTemplate.send("annotated36", "foo".getBytes(), "bar"); assertThat(this.listener.keyLatch.await(30, TimeUnit.SECONDS)).isTrue(); assertThat(this.listener.convertedKey).isEqualTo("foo"); + assertThat(this.config.intercepted).isTrue(); } @Configuration @@ -747,7 +748,11 @@ public class EnableKafkaIntegrationTests { @EnableTransactionManagement(proxyTargetClass = true) public static class Config implements KafkaListenerConfigurer { - private final CountDownLatch spyLatch = new CountDownLatch(2); + final CountDownLatch spyLatch = new CountDownLatch(2); + + volatile Throwable globalErrorThrowable; + + volatile boolean intercepted; @Bean public static PropertySourcesPlaceholderConfigurer ppc() { @@ -770,8 +775,6 @@ public class EnableKafkaIntegrationTests { return new ChainedKafkaTransactionManager<>(ktm(), transactionManager()); } - private Throwable globalErrorThrowable; - @Bean public KafkaListenerContainerFactory> kafkaListenerContainerFactory() { @@ -857,6 +860,10 @@ public class EnableKafkaIntegrationTests { ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(bytesStringConsumerFactory()); + factory.setRecordInterceptor(record -> { + this.intercepted = true; + return record; + }); return factory; } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 84ae48ae..c19ec3cf 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -112,11 +112,13 @@ public class ConcurrentMessageListenerContainerTests { ContainerProperties containerProps = new ContainerProperties(topic1); containerProps.setLogContainerConfig(true); - final CountDownLatch latch = new CountDownLatch(4); + final CountDownLatch latch = new CountDownLatch(3); final Set listenerThreadNames = new ConcurrentSkipListSet<>(); + final List payloads = new ArrayList<>(); containerProps.setMessageListener((MessageListener) message -> { ConcurrentMessageListenerContainerTests.this.logger.info("auto: " + message); listenerThreadNames.add(Thread.currentThread().getName()); + payloads.add(message.value()); latch.countDown(); }); @@ -132,6 +134,11 @@ public class ConcurrentMessageListenerContainerTests { stopLatch.countDown(); } }); + CountDownLatch intercepted = new CountDownLatch(4); + container.setRecordInterceptor(record -> { + intercepted.countDown(); + return record.value().equals("baz") ? null : record; + }); container.start(); ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic()); @@ -146,6 +153,7 @@ public class ConcurrentMessageListenerContainerTests { template.sendDefault(0, "baz"); template.sendDefault(2, "qux"); template.flush(); + assertThat(intercepted.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); for (String threadName : listenerThreadNames) { assertThat(threadName).contains("-C-"); @@ -161,6 +169,7 @@ public class ConcurrentMessageListenerContainerTests { Set> children = new HashSet<>(containers); container.stop(); assertThat(stopLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(payloads).containsExactlyInAnyOrder("foo", "bar", "qux"); events.forEach(e -> { assertThat(e.getContainer(MessageListenerContainer.class)).isSameAs(container); if (e instanceof ContainerStoppedEvent) { diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index c599e6ce..9d21e967 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -642,6 +642,10 @@ Two `MessageListenerContainer` implementations are provided: The `KafkaMessageListenerContainer` receives all message from all topics or partitions on a single thread. The `ConcurrentMessageListenerContainer` delegates to one or more `KafkaMessageListenerContainer` instances to provide multi-threaded consumption. +Starting with version 2.1.7, you can add a `RecordInterceptor` to the listener container; it will be invoked before calling the listener allowing inspection or modification of the record. +If the interceptor returns null, the listener is not called. +The interceptor is not invoked when the listener is a <>. + [[kafka-container]] ====== Using `KafkaMessageListenerContainer`