ProducerListener Improvements

Added methods receiving ProducerRecord to ProducerListener

LoggingProducerListener implements ProducerListener directly

Added test for ProducerListener with ProducerRecord

Deprecate isInterestedInSuccess for ProducerListener

Final Polishing

Author: endSly

spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java

@@ -56,6 +56,7 @@
  * @author Igor Stepanov
  * @author Artem Bilan
  * @author Biju Kunjummen
+ * @author Endika Gutiérrez
  */
 public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
 
@@ -352,10 +353,8 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
 				try {
 					if (exception == null) {
 						future.set(new SendResult<>(producerRecord, metadata));
-						if (KafkaTemplate.this.producerListener != null
-								&& KafkaTemplate.this.producerListener.isInterestedInSuccess()) {
-							KafkaTemplate.this.producerListener.onSuccess(producerRecord.topic(),
-									producerRecord.partition(), producerRecord.key(), producerRecord.value(), metadata);
+						if (KafkaTemplate.this.producerListener != null) {
+							KafkaTemplate.this.producerListener.onSuccess(producerRecord, metadata);
 						}
 						if (KafkaTemplate.this.logger.isTraceEnabled()) {
 							KafkaTemplate.this.logger.trace("Sent ok: " + producerRecord + ", metadata: " + metadata);
@@ -364,11 +363,7 @@ public void onCompletion(RecordMetadata metadata, Exception exception) {
 					else {
 						future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception));
 						if (KafkaTemplate.this.producerListener != null) {
-							KafkaTemplate.this.producerListener.onError(producerRecord.topic(),
-									producerRecord.partition(),
-									producerRecord.key(),
-									producerRecord.value(),
-									exception);
+							KafkaTemplate.this.producerListener.onError(producerRecord, exception);
 						}
 						if (KafkaTemplate.this.logger.isDebugEnabled()) {
 							KafkaTemplate.this.logger.debug("Failed to send: " + producerRecord, exception);

spring-kafka/src/main/java/org/springframework/kafka/support/LoggingProducerListener.java

@@ -30,7 +30,7 @@
  * @author Marius Bogoevici
  * @author Gary Russell
  */
-public class LoggingProducerListener<K, V> extends ProducerListenerAdapter<K, V> {
+public class LoggingProducerListener<K, V> implements ProducerListener<K, V> {
 
 	private static final Log log = LogFactory.getLog(LoggingProducerListener.class);
 

spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java

@@ -16,6 +16,7 @@
 
 package org.springframework.kafka.support;
 
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 
 /**
@@ -30,35 +31,64 @@
  *
  * @author Marius Bogoevici
  * @author Gary Russell
+ * @author Endika Gutiérrez
  *
  * @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 producerRecord the actual sent record
+	 * @param recordMetadata the result of the successful send operation
+	 */
+	default void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata) {
+		onSuccess(producerRecord.topic(), producerRecord.partition(),
+				producerRecord.key(), producerRecord.value(), recordMetadata);
+	}
+
+	/**
+	 * Invoked after the successful send of a message (that is, after it has been acknowledged by the broker).
+	 * If the method receiving the ProducerRecord is overridden, this method won't be called
 	 * @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);
+	default void onSuccess(String topic, Integer partition, K key, V value, RecordMetadata recordMetadata) {
+	}
+
+	/**
+	 * Invoked after an attempt to send a message has failed.
+	 * @param producerRecord the failed record
+	 * @param exception the exception thrown
+	 */
+	default void onError(ProducerRecord<K, V> producerRecord, Exception exception) {
+		onError(producerRecord.topic(), producerRecord.partition(),
+				producerRecord.key(), producerRecord.value(), exception);
+	}
 
 	/**
 	 * Invoked after an attempt to send a message has failed.
+	 * If the method receiving the ProducerRecord is overridden, this method won't be called
 	 * @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);
+	default void onError(String topic, Integer partition, K key, V value, Exception exception) {
+	}
 
 	/**
 	 * Return true if this listener is interested in success as well as failure.
+	 * @deprecated the result of this method will be ignored.
 	 * @return true to express interest in successful sends.
 	 */
-	boolean isInterestedInSuccess();
+	@Deprecated
+	default boolean isInterestedInSuccess() {
+		return false;
+	}
 
 }

spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java

@@ -16,31 +16,19 @@
 
 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.
  *
+ * @deprecated as the {@link ProducerListener} has default methods and can be implemented directly without the need for this adapter
+ *
  * @param <K> the key type.
  * @param <V> the value type.
  *
  * @author Marius Bogoevici
  * @author Gary Russell
  * @author Artem Bilan
+ * @author Endika Gutiérrez
  */
+@Deprecated
 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) {
-	}
-
-	@Override
-	public boolean isInterestedInSuccess() {
-		return false;
-	}
-
 }

spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java

@@ -34,6 +34,7 @@
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
@@ -50,7 +51,7 @@
 import org.springframework.kafka.support.Acknowledgment;
 import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
 import org.springframework.kafka.support.KafkaHeaders;
-import org.springframework.kafka.support.ProducerListenerAdapter;
+import org.springframework.kafka.support.ProducerListener;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.kafka.support.converter.MessagingMessageConverter;
 import org.springframework.kafka.test.rule.KafkaEmbedded;
@@ -65,6 +66,7 @@
  * @author Artem Bilan
  * @author Igor Stepanov
  * @author Biju Kunjummen
+ * @author Endika Gutiérrez
  */
 public class KafkaTemplateTests {
 
@@ -241,17 +243,36 @@ public void withListener() throws Exception {
 		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
 		template.setDefaultTopic(INT_KEY_TOPIC);
 		final CountDownLatch latch = new CountDownLatch(1);
-		template.setProducerListener(new ProducerListenerAdapter<Integer, String>() {
+		template.setProducerListener(new ProducerListener<Integer, String>() {
 
 			@Override
 			public void onSuccess(String topic, Integer partition, Integer key, String value,
 					RecordMetadata recordMetadata) {
 				latch.countDown();
 			}
 
+		});
+		template.sendDefault("foo");
+		template.flush();
+		assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
+
+		//Drain the topic
+		KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
+		pf.destroy();
+	}
+
+	@Test
+	public void withProducerRecordListener() throws Exception {
+		Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+		DefaultKafkaProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
+		KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
+		template.setDefaultTopic(INT_KEY_TOPIC);
+		final CountDownLatch latch = new CountDownLatch(1);
+		template.setProducerListener(new ProducerListener<Integer, String>() {
+
 			@Override
-			public boolean isInterestedInSuccess() {
-				return true;
+			public void onSuccess(ProducerRecord<Integer, String> record, RecordMetadata recordMetadata) {
+				latch.countDown();
 			}
 
 		});