Add RecordMetadata to ProducerListener.onError

When sending a record with no partition, and an error occurs, the
`RecordMetadata` might contain the assigned partition if the error occurs
after partition assignment; this may be useful information for the
application.

* Add @nullable

Author: garyrussell

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

@@ -385,6 +385,7 @@ public ListenableFuture<SendResult<K, V>> send(String topic, Integer partition,
 
 	@Override
 	public ListenableFuture<SendResult<K, V>> send(ProducerRecord<K, V> record) {
+		Assert.notNull(record, "'record' cannot be null");
 		return doSend(record);
 	}
 
@@ -601,7 +602,7 @@ private Callback buildCallback(final ProducerRecord<K, V> producerRecord, final
 					}
 					future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception));
 					if (KafkaTemplate.this.producerListener != null) {
-						KafkaTemplate.this.producerListener.onError(producerRecord, exception);
+						KafkaTemplate.this.producerListener.onError(producerRecord, metadata, exception);
 					}
 					KafkaTemplate.this.logger.debug(exception, () -> "Failed to send: " + producerRecord);
 				}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2020 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.
@@ -74,8 +74,14 @@ public void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata record
 	}
 
 	@Override
+	@Deprecated
 	public void onError(ProducerRecord<K, V> producerRecord, Exception exception) {
 		this.delegates.forEach(d -> d.onError(producerRecord, exception));
 	}
 
+	@Override
+	public void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata, Exception exception) {
+		this.delegates.forEach(d -> d.onError(producerRecord, recordMetadata, exception));
+	}
+
 }

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

@@ -18,8 +18,10 @@
 
 import org.apache.commons.logging.LogFactory;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
 
 import org.springframework.core.log.LogAccessor;
+import org.springframework.lang.Nullable;
 import org.springframework.util.ObjectUtils;
 
 /**
@@ -65,7 +67,7 @@ public void setMaxContentLogged(int maxContentLogged) {
 	}
 
 	@Override
-	public void onError(ProducerRecord<K, V> record, Exception exception) {
+	public void onError(ProducerRecord<K, V> record, @Nullable RecordMetadata recordMetadata, Exception exception) {
 		LOGGER.error(exception, () -> {
 			StringBuffer logOutput = new StringBuffer();
 			logOutput.append("Exception thrown when sending a message");
@@ -79,7 +81,9 @@ public void onError(ProducerRecord<K, V> record, Exception exception) {
 			}
 			logOutput.append(" to topic ").append(record.topic());
 			if (record.partition() != null) {
-				logOutput.append(" and partition ").append(record.partition());
+				logOutput.append(" and partition ").append(recordMetadata != null
+						? recordMetadata.partition()
+						: record.partition());
 			}
 			logOutput.append(":");
 			return logOutput.toString();

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2019 the original author or authors.
+ * Copyright 2015-2020 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.
@@ -19,6 +19,8 @@
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.clients.producer.RecordMetadata;
 
+import org.springframework.lang.Nullable;
+
 /**
  * Listener for handling outbound Kafka messages. Exactly one of its methods will be invoked, depending on whether
  * the write has been acknowledged or not.
@@ -49,8 +51,28 @@ default void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recor
 	 * Invoked after an attempt to send a message has failed.
 	 * @param producerRecord the failed record
 	 * @param exception the exception thrown
+	 * @deprecated in favor of {@link #onError(ProducerRecord, RecordMetadata, Exception)}.
 	 */
+	@Deprecated
 	default void onError(ProducerRecord<K, V> producerRecord, Exception exception) {
 	}
 
+	/**
+	 * Invoked after an attempt to send a message has failed.
+	 * @param producerRecord the failed record
+	 * @param recordMetadata The metadata for the record that was sent (i.e. the partition
+	 * and offset). If an error occurred, metadata will contain only valid topic and maybe
+	 * the partition. If the partition is not provided in the ProducerRecord and an error
+	 * occurs before partition is assigned, then the partition will be set to
+	 * RecordMetadata.UNKNOWN_PARTITION.
+	 * @param exception the exception thrown
+	 * @since 2.6.2
+	 */
+	@SuppressWarnings("deprecation")
+	default void onError(ProducerRecord<K, V> producerRecord, @Nullable RecordMetadata recordMetadata,
+			Exception exception) {
+
+		onError(producerRecord, exception);
+	}
+
 }

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

@@ -53,6 +53,7 @@
 import org.apache.kafka.common.Metric;
 import org.apache.kafka.common.MetricName;
 import org.apache.kafka.common.PartitionInfo;
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.errors.TimeoutException;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.serialization.Serializer;
@@ -276,7 +277,9 @@ public void onSuccess(ProducerRecord<Integer, String> record, RecordMetadata rec
 			}
 
 			@Override
-			public void onError(ProducerRecord<Integer, String> producerRecord, Exception exception) {
+			public void onError(ProducerRecord<Integer, String> producerRecord, RecordMetadata metadata,
+					Exception exception) {
+
 				assertThat(producerRecord).isNotNull();
 				assertThat(exception).isNotNull();
 				onErrorDelegateCalls.incrementAndGet();
@@ -298,7 +301,8 @@ public void onError(ProducerRecord<Integer, String> producerRecord, Exception ex
 		//Drain the topic
 		KafkaTestUtils.getSingleRecord(consumer, INT_KEY_TOPIC);
 		pf.destroy();
-		cpl.onError(records.get(0), new RuntimeException("x"));
+		cpl.onError(records.get(0), new RecordMetadata(new TopicPartition(INT_KEY_TOPIC, -1), 0L, 0L, 0L, 0L, 0, 0),
+				new RuntimeException("x"));
 		assertThat(onErrorDelegateCalls.get()).isEqualTo(2);
 	}
 

src/reference/asciidoc/kafka.adoc

@@ -309,7 +309,8 @@ public interface ProducerListener<K, V> {
 
     void onSuccess(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata);
 
-    void onError(ProducerRecord<K, V> producerRecord, Exception exception);
+    void onError(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata,
+            Exception exception);
 
 }
 ----