GH-36: Port ProducerListener from S-I-K

Resolves #36 (https://github.com/spring-projects/spring-kafka/issues/36)

Polishing according PR comments

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<K, V> implements KafkaOperations<K, V> {
 
 	private volatile String defaultTopic;
 
+	private volatile ProducerListener<K, V> producerListener;
+
 	/**
 	 * Create an instance using the supplied producer factory.
 	 * @param producerFactory the producer factory.
@@ -68,6 +73,15 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 		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<K, V> producerListener) {
+		this.producerListener = producerListener;
+	}
+
 	@Override
 	public Future<RecordMetadata>  convertAndSend(V data) {
 		return convertAndSend(this.defaultTopic, data);
@@ -163,7 +177,15 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 		if (this.logger.isTraceEnabled()) {
 			this.logger.trace("Sending: " + producerRecord);
 		}
-		Future<RecordMetadata> future = this.producer.send(producerRecord);
+		Future<RecordMetadata> 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);
 		}

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<K, V> extends ProducerListenerAdapter<K, V> {
+
+	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) + "...";
+	}
+
+}

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<K, V> {
+
+	/**
+	 * 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);
+
+}

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<K, V> implements ProducerListener<K, V> {
+
+	@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) {
+	}
+
+}

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<K, V> implements Callback {
+
+	private final String topic;
+
+	private final Integer partition;
+
+	private final K key;
+
+	private final V value;
+
+	private final ProducerListener<K, V> producerListener;
+
+	public ProducerListenerInvokingCallback(String topic, Integer partition, K key, V value,
+											ProducerListener<K, V> 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);
+		}
+	}
+
+}

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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(TEMPLATE_TOPIC);
+		final CountDownLatch latch = new CountDownLatch(1);
+		template.setProducerListener(new ProducerListenerAdapter<Integer, String>() {
+
+			@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();
+	}
+
 }