GH-694: Add recordMetadataChannel (send confirm.)

Resolves #694

Also fix deprecation of `ChannelInterceptorAware`.

docs/src/main/asciidoc/overview.adoc

@@ -332,6 +332,16 @@ Set to `true` to override the default binding destination (topic name) with the
 If the header is not present, the default binding destination is used.
 Default: `false`.
 +
+recordMetadataChannel::
+The bean name of a `MessageChannel` to which successful send results should be sent; the bean must exist in the application context.
+The message sent to the channel is the sent message (after conversion, if any) with an additional header `KafkaHeaders.RECORD_METADATA`.
+The header contains a `RecordMetadata` object provided by the Kafka client; it includes the partition and offset where the record was written in the topic.
+
+`ResultMetadata meta = sendResultMsg.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class)`
+
+Failed sends go the producer error channel (if configured); see <<kafka-error-channels>>.
+Default: null
++
 
 NOTE: The Kafka binder uses the `partitionCount` setting of the producer as a hint to create a topic with the given partition count (in conjunction with the `minPartitionCount`, the maximum of the two being the value being used).
 Exercise caution when configuring both `minPartitionCount` for a binder and `partitionCount` for an application, as the larger value is used.

spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java

@@ -52,6 +52,8 @@ public class KafkaProducerProperties {
 
 	private boolean useTopicHeader;
 
+	private String recordMetadataChannel;
+
 	public int getBufferSize() {
 		return this.bufferSize;
 	}
@@ -148,6 +150,14 @@ public class KafkaProducerProperties {
 		this.useTopicHeader = useTopicHeader;
 	}
 
+	public String getRecordMetadataChannel() {
+		return this.recordMetadataChannel;
+	}
+
+	public void setRecordMetadataChannel(String recordMetadataChannel) {
+		this.recordMetadataChannel = recordMetadataChannel;
+	}
+
 	/**
 	 * Enumeration for compression types.
 	 */

spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java

@@ -79,7 +79,6 @@ import org.springframework.expression.common.LiteralExpression;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.StaticMessageHeaderAccessor;
 import org.springframework.integration.acks.AcknowledgmentCallback;
-import org.springframework.integration.channel.ChannelInterceptorAware;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
 import org.springframework.integration.kafka.inbound.KafkaMessageSource;
@@ -112,6 +111,7 @@ import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.support.ChannelInterceptor;
 import org.springframework.messaging.support.ErrorMessage;
+import org.springframework.messaging.support.InterceptableChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
@@ -346,8 +346,8 @@ public class KafkaMessageChannelBinder extends
 						+ partitions.size()
 						+ " for the topic. The larger number will be used instead.");
 			}
-			List<ChannelInterceptor> interceptors = ((ChannelInterceptorAware) channel)
-					.getChannelInterceptors();
+			List<ChannelInterceptor> interceptors = ((InterceptableChannel) channel)
+					.getInterceptors();
 			interceptors.forEach((interceptor) -> {
 				if (interceptor instanceof PartitioningInterceptor) {
 					((PartitioningInterceptor) interceptor)
@@ -368,6 +368,9 @@ public class KafkaMessageChannelBinder extends
 		if (errorChannel != null) {
 			handler.setSendFailureChannel(errorChannel);
 		}
+		if (StringUtils.hasText(producerProperties.getExtension().getRecordMetadataChannel())) {
+			handler.setSendSuccessChannelName(producerProperties.getExtension().getRecordMetadataChannel());
+		}
 		KafkaHeaderMapper mapper = null;
 		if (this.configurationProperties.getHeaderMapperBeanName() != null) {
 			mapper = getApplicationContext().getBean(

spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java

@@ -49,6 +49,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.KafkaFuture;
 import org.apache.kafka.common.errors.TopicExistsException;
 import org.apache.kafka.common.record.TimestampType;
@@ -3111,6 +3112,42 @@ public class KafkaBinderTests extends
 		}
 	}
 
+	@Test
+	@SuppressWarnings("unchecked")
+	public void testRecordMetadata() throws Exception {
+		Binding<?> producerBinding = null;
+		try {
+			String testPayload = "test";
+
+			ExtendedProducerProperties<KafkaProducerProperties> producerProperties = createProducerProperties();
+			producerProperties.getExtension().setRecordMetadataChannel("metaChannel");
+			QueueChannel metaChannel = new QueueChannel();
+
+			DirectChannel moduleOutputChannel = createBindableChannel("output",
+					createProducerBindingProperties(producerProperties));
+
+			String testTopicName = "existing" + System.currentTimeMillis();
+			KafkaTestBinder binder = getBinder();
+			((GenericApplicationContext) binder.getApplicationContext()).registerBean("metaChannel",
+					MessageChannel.class, () -> metaChannel);
+			producerBinding = binder.bindProducer(testTopicName, moduleOutputChannel,
+					producerProperties);
+			moduleOutputChannel
+					.send(new GenericMessage<>("foo", Collections.singletonMap(KafkaHeaders.PARTITION_ID, 0)));
+			Message<?> sendResult = metaChannel.receive(10_000);
+			assertThat(sendResult).isNotNull();
+			RecordMetadata meta = sendResult.getHeaders().get(KafkaHeaders.RECORD_METADATA, RecordMetadata.class);
+			assertThat(meta).isNotNull()
+				.hasFieldOrPropertyWithValue("topic", testTopicName)
+				.hasFieldOrPropertyWithValue("partition", 0)
+				.hasFieldOrPropertyWithValue("offset", 0L);
+		}
+		finally {
+			if (producerBinding != null) {
+				producerBinding.unbind();
+			}
+		}
+	}
 
 	private final class FailingInvocationCountingMessageHandler
 			implements MessageHandler {