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); + } +}