diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 140905477..e207522fe 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/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 <>. +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. diff --git a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java b/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java index 491ac8f98..0e1db7204 100644 --- a/spring-cloud-stream-binder-kafka-core/src/main/java/org/springframework/cloud/stream/binder/kafka/properties/KafkaProducerProperties.java +++ b/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. */ diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 940f77673..248c8f25f 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/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 interceptors = ((ChannelInterceptorAware) channel) - .getChannelInterceptors(); + List 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( diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index 6c888fc3f..9fd6a594c 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/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 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 {