diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java index f3e20eaa66..1831a53e7d 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java @@ -18,8 +18,11 @@ package org.springframework.integration.kafka.dsl; import java.util.Collections; import java.util.Map; +import java.util.function.BiConsumer; import java.util.function.Consumer; +import org.apache.kafka.common.TopicPartition; + import org.springframework.integration.dsl.ComponentsRegistration; import org.springframework.integration.dsl.MessagingGatewaySpec; import org.springframework.integration.kafka.inbound.KafkaInboundGateway; @@ -27,6 +30,7 @@ import org.springframework.integration.support.ObjectStringMapBuilder; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; +import org.springframework.kafka.listener.ConsumerSeekAware; import org.springframework.kafka.support.converter.RecordMessageConverter; import org.springframework.retry.RecoveryCallback; import org.springframework.retry.support.RetryTemplate; @@ -41,6 +45,7 @@ import org.springframework.util.Assert; * @param the target {@link KafkaInboundGatewaySpec} implementation type. * * @author Gary Russell + * @author Artem Bilan * * @since 3.0.2 */ @@ -91,6 +96,20 @@ public class KafkaInboundGatewaySpec, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) { + this.target.setOnPartitionsAssignedSeekCallback(onPartitionsAssignedCallback); + return _this(); + } + @Override public Map getComponentsToRegister() { return Collections.singletonMap(this.container, getId() == null ? null : getId() + ".container"); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java index 6a9a480dd5..048871a037 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java @@ -18,13 +18,17 @@ package org.springframework.integration.kafka.dsl; import java.util.Collections; import java.util.Map; +import java.util.function.BiConsumer; import java.util.function.Consumer; +import org.apache.kafka.common.TopicPartition; + import org.springframework.integration.dsl.ComponentsRegistration; import org.springframework.integration.dsl.MessageProducerSpec; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; +import org.springframework.kafka.listener.ConsumerSeekAware; import org.springframework.kafka.listener.adapter.RecordFilterStrategy; import org.springframework.kafka.support.converter.BatchMessageConverter; import org.springframework.kafka.support.converter.MessageConverter; @@ -156,6 +160,20 @@ public class KafkaMessageDrivenChannelAdapterSpec, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) { + this.target.setOnPartitionsAssignedSeekCallback(onPartitionsAssignedCallback); + return _this(); + } + @Override public Map getComponentsToRegister() { return Collections.singletonMap(this.container, getId() == null ? null : getId() + ".container"); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index 66975f4ba5..d1543383da 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -16,10 +16,13 @@ package org.springframework.integration.kafka.inbound; +import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; import org.springframework.core.AttributeAccessor; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -32,6 +35,7 @@ import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.listener.AbstractMessageListenerContainer; +import org.springframework.kafka.listener.ConsumerSeekAware; import org.springframework.kafka.listener.MessageListener; import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter; @@ -58,6 +62,7 @@ import org.springframework.util.Assert; * @param the reply value type. * * @author Gary Russell + * @author Artem Bilan * * @since 3.0.2 * @@ -76,6 +81,8 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem private RecoveryCallback recoveryCallback; + private BiConsumer, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedSeekCallback; + /** * Construct an instance with the provided container. * @param messageListenerContainer the container. @@ -133,6 +140,21 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem this.recoveryCallback = recoveryCallback; } + /** + * Specify a {@link BiConsumer} for seeks management during + * {@link ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)} + * call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}. + * This is called from the internal + * {@link org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter} implementation. + * @param onPartitionsAssignedCallback the {@link BiConsumer} to use + * @since 3.0.4 + * @see ConsumerSeekAware#onPartitionsAssigned + */ + public void setOnPartitionsAssignedSeekCallback( + BiConsumer, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) { + this.onPartitionsAssignedSeekCallback = onPartitionsAssignedCallback; + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -212,6 +234,13 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem super(null, null); } + @Override + public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) { + if (KafkaInboundGateway.this.onPartitionsAssignedSeekCallback != null) { + KafkaInboundGateway.this.onPartitionsAssignedSeekCallback.accept(assignments, callback); + } + } + @Override public void onMessage(ConsumerRecord record, Acknowledgment acknowledgment, Consumer consumer) { Message message = null; @@ -254,7 +283,7 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1); if (message.getHeaders() instanceof KafkaMessageHeaders) { ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders() - .put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt); + .put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt); } else { messageToReturn = MessageBuilder.fromMessage(message) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index e0ad7493c8..5ea20e2dee 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -17,10 +17,13 @@ package org.springframework.integration.kafka.inbound; import java.util.List; +import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.common.TopicPartition; import org.springframework.core.AttributeAccessor; import org.springframework.integration.IntegrationMessageHeaderAccessor; @@ -33,6 +36,7 @@ import org.springframework.integration.support.ErrorMessageUtils; import org.springframework.integration.support.MessageBuilder; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.BatchMessageListener; +import org.springframework.kafka.listener.ConsumerSeekAware; import org.springframework.kafka.listener.MessageListener; import org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter; import org.springframework.kafka.listener.adapter.FilteringBatchMessageListenerAdapter; @@ -90,6 +94,8 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo private boolean filterInRetry; + private BiConsumer, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedSeekCallback; + /** * Construct an instance with mode {@link ListenerMode#record}. * @param messageListenerContainer the container. @@ -225,6 +231,21 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo this.batchListener.setFallbackType(payloadType); } + /** + * Specify a {@link BiConsumer} for seeks management during + * {@link ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)} + * call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}. + * This is called from the internal + * {@link org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter} implementation. + * @param onPartitionsAssignedCallback the {@link BiConsumer} to use + * @since 3.0.4 + * @see ConsumerSeekAware#onPartitionsAssigned + */ + public void setOnPartitionsAssignedSeekCallback( + BiConsumer, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) { + this.onPartitionsAssignedSeekCallback = onPartitionsAssignedCallback; + } + @Override public String getComponentType() { return "kafka:message-driven-channel-adapter"; @@ -342,6 +363,23 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo } } + private void sendMessageIfAny(Message message, Object kafkaConsumedObject) { + if (message != null) { + try { + sendMessage(message); + } + finally { + if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) { + attributesHolder.remove(); + } + } + } + else { + KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: " + + kafkaConsumedObject); + } + } + /** * The listener mode for the container, record or batch. * @since 1.2 @@ -368,6 +406,13 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo super(null, null); } + @Override + public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) { + if (KafkaMessageDrivenChannelAdapter.this.onPartitionsAssignedSeekCallback != null) { + KafkaMessageDrivenChannelAdapter.this.onPartitionsAssignedSeekCallback.accept(assignments, callback); + } + } + @Override public void onMessage(ConsumerRecord record, Acknowledgment acknowledgment, Consumer consumer) { Message message = null; @@ -382,20 +427,8 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo RuntimeException exception = new ConversionException("Failed to convert to message for: " + record, e); sendErrorMessageIfNecessary(null, exception); } - if (message != null) { - try { - sendMessage(message); - } - finally { - if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) { - attributesHolder.remove(); - } - } - } - else { - KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: " - + record); - } + + sendMessageIfAny(message, record); } private Message addDeliveryAttemptHeader(Message message) { @@ -404,7 +437,7 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1); if (message.getHeaders() instanceof KafkaMessageHeaders) { ((KafkaMessageHeaders) message.getHeaders()).getRawHeaders() - .put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt); + .put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt); } else { messageToReturn = MessageBuilder.fromMessage(message) @@ -444,8 +477,15 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo } @Override - public void onMessage(List> records, Acknowledgment acknowledgment, - Consumer consumer) { + public void onPartitionsAssigned(Map assignments, ConsumerSeekCallback callback) { + if (KafkaMessageDrivenChannelAdapter.this.onPartitionsAssignedSeekCallback != null) { + KafkaMessageDrivenChannelAdapter.this.onPartitionsAssignedSeekCallback.accept(assignments, callback); + } + } + + @Override + public void onMessage(List> records, Acknowledgment acknowledgment, + Consumer consumer) { Message message = null; try { @@ -458,20 +498,8 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo getMessagingTemplate().send(getErrorChannel(), new ErrorMessage(exception)); } } - if (message != null) { - try { - sendMessage(message); - } - finally { - if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) { - attributesHolder.remove(); - } - } - } - else { - KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: " - + records); - } + + sendMessageIfAny(message, records); } @Override diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 0a15a2a607..d161ac1540 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -210,6 +210,8 @@ public class KafkaDslTests { this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo"); assertThat(this.config.sourceFlowLatch.await(10, TimeUnit.SECONDS)).isTrue(); assertThat(this.config.fromSource).isEqualTo("foo"); + + assertThat(this.config.onPartitionsAssignedCalledLatch.await(10, TimeUnit.SECONDS)).isTrue(); } @Test @@ -227,6 +229,8 @@ public class KafkaDslTests { private final CountDownLatch replyContainerLatch = new CountDownLatch(1); + private final CountDownLatch onPartitionsAssignedCalledLatch = new CountDownLatch(1); + private Object fromSource; @Bean @@ -249,11 +253,14 @@ public class KafkaDslTests { KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1) .configureListenerContainer(c -> c.ackMode(ContainerProperties.AckMode.MANUAL) + .idleEventInterval(100L) .id("topic1ListenerContainer")) .recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(), new RawRecordHeaderErrorMessageStrategy())) .retryTemplate(new RetryTemplate()) - .filterInRetry(true)) + .filterInRetry(true) + .onPartitionsAssignedSeekCallback((map, callback) -> + ContextConfiguration.this.onPartitionsAssignedCalledLatch.countDown())) .filter(Message.class, m -> m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101, f -> f.throwExceptionOnRejection(true)) diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java index cea31b7834..3478ba0574 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java @@ -23,6 +23,8 @@ import static org.springframework.kafka.test.assertj.KafkaConditions.value; import java.lang.reflect.Type; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerConfig; @@ -105,6 +107,7 @@ public class InboundGatewayTests { props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); + containerProps.setIdleEventInterval(100L); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); @@ -133,6 +136,11 @@ public class InboundGatewayTests { } }); + + CountDownLatch onPartitionsAssignedCalledLatch = new CountDownLatch(1); + + gateway.setOnPartitionsAssignedSeekCallback((map, seekConsumer) -> onPartitionsAssignedCalledLatch.countDown()); + gateway.start(); ContainerTestUtils.waitForAssignment(container, 2); @@ -149,11 +157,13 @@ public class InboundGatewayTests { assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME"); assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic2); assertThat(headers.get("testHeader")).isEqualTo("testValue"); + reply.send(MessageBuilder.withPayload("FOO").copyHeaders(headers).build()); ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic2); assertThat(record).has(partition(1)); assertThat(record).has(value("FOO")); + assertThat(onPartitionsAssignedCalledLatch.await(10, TimeUnit.SECONDS)).isTrue(); gateway.stop(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index 465f89e759..f385e2b5a0 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -302,12 +302,18 @@ public class MessageDrivenAdapterTests { props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic2); + containerProps.setIdleEventInterval(100L); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); KafkaMessageDrivenChannelAdapter adapter = new KafkaMessageDrivenChannelAdapter<>(container, ListenerMode.batch); QueueChannel out = new QueueChannel(); adapter.setOutputChannel(out); + + final CountDownLatch onPartitionsAssignedCalledLatch = new CountDownLatch(1); + + adapter.setOnPartitionsAssignedSeekCallback((map, consumer) -> onPartitionsAssignedCalledLatch.countDown()); + adapter.afterPropertiesSet(); adapter.setBatchMessageConverter(new BatchMessagingMessageConverter() { @@ -347,6 +353,8 @@ public class MessageDrivenAdapterTests { .isEqualTo(Arrays.asList(1487694048607L, 1487694048608L)); assertThat(headers.get("testHeader")).isEqualTo("testValue"); + assertThat(onPartitionsAssignedCalledLatch.await(10, TimeUnit.SECONDS)).isTrue(); + adapter.setMessageConverter(new BatchMessageConverter() { @Override diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 43863a5087..6e8cb3b3a0 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -123,8 +123,8 @@ public class KafkaProducerMessageHandlerTests { } @Test - public void testOutbound() { - ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + public void testOutbound() throws Exception { + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); @@ -174,11 +174,13 @@ public class KafkaProducerMessageHandlerTests { assertThat(record).has(key(2)); assertThat(record).has(partition(1)); assertThat(record.value()).isNull(); + + producerFactory.destroy(); } @Test - public void testOutboundWithTimestamp() { - ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + public void testOutboundWithTimestamp() throws Exception { + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); @@ -203,11 +205,13 @@ public class KafkaProducerMessageHandlerTests { new DefaultKafkaHeaderMapper().toHeaders(record.headers(), headers); assertThat(headers.size()).isEqualTo(1); assertThat(headers.get("baz")).isEqualTo("qux"); + + producerFactory.destroy(); } @Test - public void testOutboundWithTimestampExpression() { - ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + public void testOutboundWithTimestampExpression() throws Exception { + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); @@ -240,11 +244,13 @@ public class KafkaProducerMessageHandlerTests { assertThat(record2).has(partition(1)); assertThat(record2).has(value("foo")); assertThat(record2.timestamp()).isGreaterThanOrEqualTo(currentTimeMarker); + + producerFactory.destroy(); } @Test - public void testOutboundWithAsyncResults() { - ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + public void testOutboundWithAsyncResults() throws Exception { + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); KafkaTemplate template = new KafkaTemplate<>(producerFactory); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); @@ -298,6 +304,8 @@ public class KafkaProducerMessageHandlerTests { assertThat(((MessagingException) received.getPayload()).getFailedMessage()).isSameAs(message); assertThat(((MessagingException) received.getPayload()).getCause()).isSameAs(fooException); assertThat(((KafkaSendFailureException) received.getPayload()).getRecord()).isNotNull(); + + producerFactory.destroy(); } @Test @@ -322,7 +330,7 @@ public class KafkaProducerMessageHandlerTests { KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(consumerFactory, containerProperties); - ProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( KafkaTestUtils.producerProps(embeddedKafka)); ReplyingKafkaTemplate template = new ReplyingKafkaTemplate<>(producerFactory, container); template.start(); @@ -390,6 +398,8 @@ public class KafkaProducerMessageHandlerTests { template.stop(); // discard from the test consumer KafkaTestUtils.getSingleRecord(consumer, topic6); + + producerFactory.destroy(); } @SuppressWarnings({ "rawtypes", "unchecked" })