diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index 2f6b88bb..ec8a8ff9 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java @@ -25,6 +25,9 @@ import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; +import org.springframework.kafka.support.ProducerListener; +import org.springframework.kafka.support.ProducerListenerInvokingCallback; + /** * A template for executing high-level operations. @@ -42,6 +45,8 @@ public class KafkaTemplate implements KafkaOperations { private volatile String defaultTopic; + private volatile ProducerListener producerListener; + /** * Create an instance using the supplied producer factory. * @param producerFactory the producer factory. @@ -68,6 +73,15 @@ public class KafkaTemplate implements KafkaOperations { this.defaultTopic = defaultTopic; } + /** + * Set a {@link ProducerListener} which will be invoked when Kafka acknowledges + * a send operation. + * @param producerListener the listener. + */ + public void setProducerListener(ProducerListener producerListener) { + this.producerListener = producerListener; + } + @Override public Future convertAndSend(V data) { return convertAndSend(this.defaultTopic, data); @@ -163,7 +177,15 @@ public class KafkaTemplate implements KafkaOperations { if (this.logger.isTraceEnabled()) { this.logger.trace("Sending: " + producerRecord); } - Future future = this.producer.send(producerRecord); + Future future; + if (this.producerListener == null) { + future = this.producer.send(producerRecord); + } + else { + future = this.producer.send(producerRecord, + new ProducerListenerInvokingCallback<>(producerRecord.topic(), producerRecord.partition(), + producerRecord.key(), producerRecord.value(), this.producerListener)); + } if (this.logger.isTraceEnabled()) { this.logger.trace("Sent: " + producerRecord); } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java new file mode 100644 index 00000000..cee3d03f --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java @@ -0,0 +1,84 @@ +/* + * Copyright 2015-2016 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.kafka.support; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.util.ObjectUtils; + +/** + * The {@link ProducerListener} that logs exceptions thrown when sending messages. + * + * @author Marius Bogoevici + * @author Gary Russell + */ +public class LoggingProducerListener extends ProducerListenerAdapter { + + 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, K key, V value, Exception exception) { + if (log.isErrorEnabled()) { + StringBuffer logOutput = new StringBuffer(); + logOutput.append("Exception thrown when sending a message"); + if (this.includeContents) { + logOutput.append(" with key='" + + toDisplayString(ObjectUtils.nullSafeToString(key), this.maxContentLogged) + "'"); + logOutput.append(" and payload='" + + toDisplayString(ObjectUtils.nullSafeToString(value), this.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-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java new file mode 100644 index 00000000..8309ec5f --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java @@ -0,0 +1,55 @@ +/* + * Copyright 2015-2016 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.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. + * + * @author Marius Bogoevici + * @author Gary Russell + * + * @see org.apache.kafka.clients.producer.Callback + */ +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, K key, V 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, K key, V value, Exception exception); + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java new file mode 100644 index 00000000..2af460f7 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java @@ -0,0 +1,38 @@ +/* + * Copyright 2015-2016 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.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 + * @author Gary Russell + * @author Artem Bilan + */ +public abstract class ProducerListenerAdapter implements ProducerListener { + + @Override + public void onSuccess(String topic, Integer partition, K key, V value, RecordMetadata recordMetadata) { + } + + @Override + public void onError(String topic, Integer partition, K key, V value, Exception exception) { + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerInvokingCallback.java b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerInvokingCallback.java new file mode 100644 index 00000000..bc663bd8 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerInvokingCallback.java @@ -0,0 +1,63 @@ +/* + * Copyright 2015-2016 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.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 + * @author Gary Russell + */ +public class ProducerListenerInvokingCallback implements Callback { + + private final String topic; + + private final Integer partition; + + private final K key; + + private final V value; + + private final ProducerListener producerListener; + + public ProducerListenerInvokingCallback(String topic, Integer partition, K key, V value, + ProducerListener producerListener) { + Assert.notNull(producerListener, "must not be null"); + this.topic = topic; + this.partition = partition; + this.key = key; + this.value = value; + this.producerListener = producerListener; + } + + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + if (exception != null) { + this.producerListener.onError(this.topic, this.partition, this.key, this.value, exception); + } + else { + this.producerListener.onSuccess(this.topic, this.partition, this.key, this.value, metadata); + } + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java index 8180d412..ce7faced 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java @@ -23,16 +23,19 @@ import static org.springframework.kafka.test.assertj.KafkaConditions.value; import java.util.Map; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; import org.junit.ClassRule; import org.junit.Test; import org.springframework.kafka.listener.ContainerTestUtils; import org.springframework.kafka.listener.KafkaMessageListenerContainer; import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.support.ProducerListenerAdapter; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; @@ -87,4 +90,25 @@ public class KafkaTemplateTests { assertThat(received).has(value("baz")); } + @Test + public void withListener() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf); + template.setDefaultTopic(TEMPLATE_TOPIC); + final CountDownLatch latch = new CountDownLatch(1); + template.setProducerListener(new ProducerListenerAdapter() { + + @Override + public void onSuccess(String topic, Integer partition, Integer key, String value, + RecordMetadata recordMetadata) { + latch.countDown(); + } + + }); + template.syncConvertAndSend("foo"); + template.flush(); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + } + }