From 464ca70aa612fbe8b3b290052b6db88a5ad7c355 Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Sat, 24 Oct 2015 16:09:09 -0400 Subject: [PATCH] Add Support for ProducerListener Introduces the concept of ProducerListener, that can be injected in a ProducerContext and receive notifications when a message has been acknowledged or rejected. While the Kafka API currently supports providing a Callback, it requires the creation of a distinct instance for receiving contextual information about the message that has been sent or rejected. This allows the use of the Callback mechanism with an injectable ProducerListener strategy. Polishing and Callback Integration Test --- .../xml/KafkaProducerContextParser.java | 2 + .../support/DefaultProducerListener.java | 36 ++++++ .../support/LoggingProducerListener.java | 83 ++++++++++++++ .../kafka/support/ProducerConfiguration.java | 19 +++- .../kafka/support/ProducerListener.java | 55 +++++++++ .../ProducerListenerInvokingCallback.java | 61 ++++++++++ .../config/spring-integration-kafka-1.3.xsd | 12 ++ ...afkaProducerContextParserTests-context.xml | 3 + .../xml/KafkaProducerContextParserTests.java | 9 +- .../kafka/outbound/OutboundTests.java | 39 ++++++- .../kafka/outbound/ProducerListenerTests.java | 105 ++++++++++++++++++ 11 files changed, 417 insertions(+), 7 deletions(-) create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultProducerListener.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/LoggingProducerListener.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListener.java create mode 100644 spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListenerInvokingCallback.java create mode 100644 spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/ProducerListenerTests.java diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java index a4435e8aa9..46dce6e4a8 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParser.java @@ -157,6 +157,8 @@ public class KafkaProducerContextParser extends AbstractSimpleBeanDefinitionPars .addConstructorArgValue(producerFactoryBeanDefinition); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(producerConfigurationBuilder, producerConfiguration, "conversion-service"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(producerConfigurationBuilder, producerConfiguration, + "producer-listener"); producerConfigurationsMap.put(producerConfiguration.getAttribute("topic"), producerConfigurationBuilder.getBeanDefinition()); } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultProducerListener.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultProducerListener.java new file mode 100644 index 0000000000..184b54c515 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/DefaultProducerListener.java @@ -0,0 +1,36 @@ +/* + * Copyright 2015 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 + * + * http://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.integration.kafka.support; + +import org.apache.kafka.clients.producer.RecordMetadata; + +/** + * No-op implementation of @link ProducerListener}, to be used as base class for other implementations. + * + * @author Marius Bogoevici + * @since 1.3 + */ +public class DefaultProducerListener implements ProducerListener { + + @Override + public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) { + } + + @Override + public void onError(String topic, Integer partition, Object key, Object value, Exception exception) { + } +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/LoggingProducerListener.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/LoggingProducerListener.java new file mode 100644 index 0000000000..4957b08d55 --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/LoggingProducerListener.java @@ -0,0 +1,83 @@ +/* + * Copyright 2015 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 + * + * http://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.integration.kafka.support; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.util.ObjectUtils; + +/** + * {@link ProducerListener} that logs exceptions thrown when sending messages. + * + * @author Marius Bogoevici + * @since 1.3 + */ +public class LoggingProducerListener extends DefaultProducerListener { + + private static final Log log = LogFactory.getLog(LoggingProducerListener.class); + + private boolean includeContents = true; + + private int maxContentLogged = 100; + + /** + * Whether the log message should include the contents (key and payload). + * + * @param includeContents true if the contents of the message should be logged + */ + public void setIncludeContents(boolean includeContents) { + this.includeContents = includeContents; + } + + /** + * The maximum amount of data to be logged for either key or password. As message sizes may vary and + * become fairly large, this allows limiting the amount of data sent to logs. + * + * @param maxContentLogged the maximum amount of data being logged. + */ + public void setMaxContentLogged(int maxContentLogged) { + this.maxContentLogged = maxContentLogged; + } + + @Override + public void onError(String topic, Integer partition, Object key, Object payload, Exception exception) { + if (log.isErrorEnabled()) { + StringBuffer logOutput = new StringBuffer(); + logOutput.append("Exception thrown when sending a message"); + if (includeContents) { + logOutput.append(" with key='" + + toDisplayString(ObjectUtils.nullSafeToString(key), maxContentLogged) + "'"); + logOutput.append(" and payload='" + + toDisplayString(ObjectUtils.nullSafeToString(payload),maxContentLogged) + "'"); + } + logOutput.append(" to topic " + topic); + if (partition != null) { + logOutput.append(" and partition " +partition); + } + logOutput.append(":"); + log.error(logOutput, exception); + } + } + + private String toDisplayString(String original, int maxCharacters) { + if (original.length() <= maxCharacters) { + return original; + } + return original.substring(0, maxCharacters) + "..."; + } +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java index cb5baf86c5..6c218c754a 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerConfiguration.java @@ -49,6 +49,8 @@ public class ProducerConfiguration { private ConversionService conversionService; + private ProducerListener producerListener; + public ProducerConfiguration(ProducerMetadata producerMetadata, Producer producer) { Assert.notNull(producerMetadata); Assert.notNull(producer); @@ -64,6 +66,10 @@ public class ProducerConfiguration { this.conversionService = conversionService; } + public void setProducerListener(ProducerListener producerListener) { + this.producerListener = producerListener; + } + public ProducerMetadata getProducerMetadata() { return this.producerMetadata; } @@ -85,9 +91,18 @@ public class ProducerConfiguration { partition = this.getProducerMetadata().getPartitioner().partition(messageKey, this.producer.partitionsFor(targetTopic).size()); } - Future future = - this.producer.send(new ProducerRecord<>(targetTopic, partition, messageKey, messagePayload)); + ProducerRecord record = new ProducerRecord<>(targetTopic, partition, messageKey, messagePayload); + Future future; + if (producerListener == null) { + future = this.producer.send(record); + } + else { + ProducerListenerInvokingCallback callback = + new ProducerListenerInvokingCallback(targetTopic, partition, messageKey, messagePayload, + producerListener); + future = this.producer.send(record, callback); + } if (!producerMetadata.isSync()) { return future; } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListener.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListener.java new file mode 100644 index 0000000000..6b29a9ea3a --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListener.java @@ -0,0 +1,55 @@ +/* + * Copyright 2015 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 + * + * http://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.integration.kafka.support; + +import org.apache.kafka.clients.producer.RecordMetadata; + +/** + * Listener for handling outbound Kafka messages. Exactly one of its methods will be invoked, depending on whether + * the write has been acknowledged or not. + * + * Its main goal is to provide a stateless singleton delegate for {@link org.apache.kafka.clients.producer.Callback}s, + * which, in all but the most trivial cases, requires creating a separate instance per message. + * + * @see org.apache.kafka.clients.producer.Callback + * @author Marius Bogoevici + * @since 1.3 + */ +public interface ProducerListener { + + /** + * Invoked after the successful send of a message (that is, after it has been acknowledged by the broker) + * + * @param topic the destination topic + * @param partition the destination partition (could be null) + * @param key the key of the outbound message + * @param value the payload of the outbound message + * @param recordMetadata the result of the successful send operation + */ + void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata); + + /** + * Invoked after an attempt to send a message has failed + * + * @param topic the destination topic + * @param partition the destination partition (could be null) + * @param key the key of the outbound message + * @param value the payload of the outbound message + * @param exception the exception thrown + */ + void onError(String topic, Integer partition, Object key, Object value,Exception exception); +} diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListenerInvokingCallback.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListenerInvokingCallback.java new file mode 100644 index 0000000000..59cb267f4d --- /dev/null +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ProducerListenerInvokingCallback.java @@ -0,0 +1,61 @@ +/* + * Copyright 2015 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 + * + * http://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.integration.kafka.support; + +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.RecordMetadata; + +import org.springframework.util.Assert; + +/** + * Adapts the {@link org.apache.kafka.clients.producer.Callback} interface of the + * {@link org.apache.kafka.clients.producer.Producer} to a {@link ProducerListener}. + * + * @author Marius Bogoevici + */ +public class ProducerListenerInvokingCallback implements Callback { + + private final String topic; + + private final Integer partition; + + private final Object key; + + private final Object payload; + + private final ProducerListener producerListener; + + public ProducerListenerInvokingCallback(String topic, Integer partition, Object key, Object payload, + ProducerListener producerListener) { + Assert.notNull(producerListener, "must not be null"); + this.topic = topic; + this.partition = partition; + this.key = key; + this.payload = payload; + this.producerListener = producerListener; + } + + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + if (exception != null) { + producerListener.onError(topic,partition,key, payload,exception); + } + else { + producerListener.onSuccess(topic, partition, key, payload, metadata); + } + } +} diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd index c63a78b77c..a43b070734 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/kafka/config/spring-integration-kafka-1.3.xsd @@ -189,6 +189,18 @@ + + + + + Listener for notifying when a message has been sent or has failed. + + + + + + + @@ -54,6 +55,8 @@ + + diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParserTests.java index 3fda1143b9..013938f238 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaProducerContextParserTests.java @@ -36,9 +36,12 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.TypeDescriptor; +import org.springframework.integration.kafka.rule.KafkaEmbedded; +import org.springframework.integration.kafka.rule.KafkaRule; import org.springframework.integration.kafka.rule.KafkaRunning; import org.springframework.integration.kafka.support.KafkaProducerContext; import org.springframework.integration.kafka.support.ProducerConfiguration; +import org.springframework.integration.kafka.support.ProducerListener; import org.springframework.integration.kafka.support.ProducerMetadata; import org.springframework.integration.kafka.util.EncoderAdaptingSerializer; import org.springframework.integration.test.util.TestUtils; @@ -55,7 +58,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; public class KafkaProducerContextParserTests { @ClassRule - public static KafkaRunning kafkaRunning = KafkaRunning.isRunning(); + public static KafkaRule kafkaRule = new KafkaEmbedded(1); @Autowired private ApplicationContext appContext; @@ -108,6 +111,10 @@ public class KafkaProducerContextParserTests { ConversionService configuredConversionService = (ConversionService) directFieldAccessor2.getPropertyValue("conversionService"); assertSame(conversionService, configuredConversionService); + final ProducerListener producerListener = appContext.getBean("producerListener", ProducerListener.class); + ProducerListener configuredProducerListener = (ProducerListener) directFieldAccessor2.getPropertyValue("producerListener"); + assertSame(producerListener, configuredProducerListener); + assertEquals(9876,producerConfigurationTest2.getProducerMetadata().getBatchBytes()); assertFalse(TestUtils.getPropertyValue(producerContext, "autoStartup", Boolean.class)); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java index 79671217d3..df106e0116 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/OutboundTests.java @@ -22,6 +22,7 @@ import static org.hamcrest.Matchers.equalTo; import static org.hamcrest.Matchers.hasItem; import static org.hamcrest.Matchers.hasSize; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import java.util.ArrayList; @@ -33,6 +34,7 @@ import java.util.UUID; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import org.apache.kafka.clients.producer.RecordMetadata; import org.junit.After; import org.junit.Rule; import org.junit.Test; @@ -53,6 +55,7 @@ import org.springframework.integration.kafka.support.KafkaHeaders; import org.springframework.integration.kafka.support.KafkaProducerContext; import org.springframework.integration.kafka.support.ProducerConfiguration; import org.springframework.integration.kafka.support.ProducerFactoryBean; +import org.springframework.integration.kafka.support.ProducerListener; import org.springframework.integration.kafka.support.ProducerMetadata; import org.springframework.integration.kafka.support.ZookeeperConnect; import org.springframework.integration.kafka.util.EncoderAdaptingSerializer; @@ -89,6 +92,7 @@ public class OutboundTests { public void tearDown() { try { AdminUtils.deleteTopic(kafkaRule.getZkClient(), TOPIC); + AdminUtils.deleteTopic(kafkaRule.getZkClient(), TOPIC2); } catch (Exception e) { } @@ -156,7 +160,7 @@ public class OutboundTests { } @Test - public void testHeaderRouting() throws Exception { + public void testHeaderRoutingAndAsyncCallback() throws Exception { // create the topic @@ -196,7 +200,24 @@ public class OutboundTests { kafkaMessageListenerContainer.start(); - KafkaProducerContext producerContext = createProducerContext(); + int expectedDeliveryConfirmations = 4; + final List results = new ArrayList(); + final CountDownLatch sendResultLatch = new CountDownLatch(expectedDeliveryConfirmations); + ProducerListener listener = new ProducerListener() { + + @Override + public void onSuccess(String topic, Integer partition, Object key, Object value, RecordMetadata recordMetadata) { + results.add(recordMetadata); + sendResultLatch.countDown(); + } + + @Override + public void onError(String topic, Integer partition, Object key, Object value, Exception exception) { + sendResultLatch.countDown(); + } + }; + + KafkaProducerContext producerContext = createProducerContext(listener); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(producerContext); @@ -226,9 +247,12 @@ public class OutboundTests { .setHeader("bar", TOPIC2) .build()); + assertTrue(sendResultLatch.await(10, TimeUnit.SECONDS)); + assertThat(results.size(), equalTo(expectedDeliveryConfirmations)); + producerContext.stop(); - latch.await(1000, TimeUnit.MILLISECONDS); + latch.await(10000, TimeUnit.MILLISECONDS); assertThat(latch.getCount(), equalTo(0L)); // messages are routed to both topics assertThat(payloadsByTopic.keysView(), hasItem(TOPIC)); @@ -309,17 +333,24 @@ public class OutboundTests { } private KafkaProducerContext createProducerContext() throws Exception { + return createProducerContext(null); + } + + private KafkaProducerContext createProducerContext(ProducerListener producerListener) throws Exception { KafkaProducerContext kafkaProducerContext = new KafkaProducerContext(); Encoder encoder = new StringEncoder(); ProducerMetadata producerMetadata = new ProducerMetadata(TOPIC, String.class, String.class, new EncoderAdaptingSerializer<>(encoder), new EncoderAdaptingSerializer<>(encoder)); Properties props = new Properties(); - props.put("linger.ms", "15000"); + if (producerListener == null) { + props.put("linger.ms", "15000"); + } ProducerFactoryBean producer = new ProducerFactoryBean<>(producerMetadata, kafkaRule.getBrokersAsString(), props); ProducerConfiguration config = new ProducerConfiguration<>(producerMetadata, producer.getObject()); + config.setProducerListener(producerListener); Map> producerConfigurationMap = Collections.>singletonMap(TOPIC, config); kafkaProducerContext.setProducerConfigurations(producerConfigurationMap); diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/ProducerListenerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/ProducerListenerTests.java new file mode 100644 index 0000000000..5d43e61274 --- /dev/null +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/ProducerListenerTests.java @@ -0,0 +1,105 @@ +/* + * Copyright 2015 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 + * + * http://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.integration.kafka.outbound; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoMoreInteractions; + +import java.util.Collections; +import java.util.Map; + +import org.apache.kafka.clients.producer.Callback; +import org.apache.kafka.clients.producer.Producer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.serialization.Serializer; +import org.hamcrest.CoreMatchers; +import org.junit.Test; +import org.mockito.ArgumentCaptor; +import org.mockito.Mockito; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.integration.kafka.support.KafkaHeaders; +import org.springframework.integration.kafka.support.KafkaProducerContext; +import org.springframework.integration.kafka.support.ProducerConfiguration; +import org.springframework.integration.kafka.support.ProducerListener; +import org.springframework.integration.kafka.support.ProducerListenerInvokingCallback; +import org.springframework.integration.kafka.support.ProducerMetadata; +import org.springframework.messaging.support.MessageBuilder; + +/** + * @author Marius Bogoevici + */ +public class ProducerListenerTests { + + @Test + @SuppressWarnings({"unchecked","rawtypes"}) + public void testProducerListenerSet() throws Exception { + KafkaProducerContext producerContext = new KafkaProducerContext(); + Serializer serializer = Mockito.mock(Serializer.class); + ProducerMetadata producerMetadata + = new ProducerMetadata<>("default", Object.class, Object.class, serializer, serializer); + Producer producer = Mockito.mock(Producer.class); + ProducerConfiguration producerConfiguration + = new ProducerConfiguration<>(producerMetadata, producer); + ProducerListener producerListener = mock(ProducerListener.class); + producerConfiguration.setProducerListener(producerListener); + Map> producerConfigurations + = Collections.>singletonMap("default", producerConfiguration); + producerContext.setProducerConfigurations(producerConfigurations); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(producerContext); + handler.handleMessage( + MessageBuilder.withPayload("somePayload") + .setHeader(KafkaHeaders.PARTITION_ID, 2) + .setHeader(KafkaHeaders.MESSAGE_KEY, "someKey") + .build()); + final ArgumentCaptor argument = ArgumentCaptor.forClass(Callback.class); + verify(producer).send(any(ProducerRecord.class), argument.capture()); + Callback callback = argument.getValue(); + assertThat(callback, CoreMatchers.instanceOf(ProducerListenerInvokingCallback.class)); + DirectFieldAccessor fieldAccessor = new DirectFieldAccessor(callback); + assertEquals(fieldAccessor.getPropertyValue("topic"),"default"); + assertEquals(fieldAccessor.getPropertyValue("partition"),2); + assertEquals(fieldAccessor.getPropertyValue("key"),"someKey"); + assertEquals(fieldAccessor.getPropertyValue("payload"),"somePayload"); + assertSame(fieldAccessor.getPropertyValue("producerListener"), producerListener); + verifyNoMoreInteractions(producer); + } + + @Test + @SuppressWarnings({"unchecked","rawtypes"}) + public void testProducerListenerNotSet() throws Exception { + KafkaProducerContext producerContext = new KafkaProducerContext(); + Serializer serializer = Mockito.mock(Serializer.class); + ProducerMetadata producerMetadata + = new ProducerMetadata<>("default", Object.class, Object.class, serializer, serializer); + Producer producer = Mockito.mock(Producer.class); + ProducerConfiguration producerConfiguration + = new ProducerConfiguration<>(producerMetadata, producer); + Map> producerConfigurations + = Collections.>singletonMap("default", producerConfiguration); + producerContext.setProducerConfigurations(producerConfigurations); + KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(producerContext); + handler.handleMessage(MessageBuilder.withPayload("somePayload").build()); + verify(producer).send(any(ProducerRecord.class)); + verifyNoMoreInteractions(producer); + } +}