diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java index 075ab51703..ae616401c9 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java @@ -59,6 +59,7 @@ import org.springframework.kafka.support.SimpleKafkaHeaderMapper; import org.springframework.kafka.support.converter.KafkaMessageHeaders; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.kafka.support.converter.RecordMessageConverter; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandlingException; @@ -72,9 +73,17 @@ import org.springframework.util.concurrent.ListenableFutureCallback; import org.springframework.util.concurrent.SettableListenableFuture; /** - * Kafka Message Handler; when supplied with a {@link ReplyingKafkaTemplate} - * it is used as the handler in an outbound gateway. When supplied with a simple - * {@link KafkaTemplate} it used as the handler in an outbound channel adapter. + * Kafka Message Handler; when supplied with a {@link ReplyingKafkaTemplate} it is used as + * the handler in an outbound gateway. When supplied with a simple {@link KafkaTemplate} + * it used as the handler in an outbound channel adapter. + *

+ * Starting with version 3.2.1 the handler supports receiving a pre-built + * {@link ProducerRecord} payload. In that case, most configuration properties + * ({@link #setTopicExpression(Expression)} etc.) are ignored. If the handler is used as + * gateway, the {@link ProducerRecord} will have its headers enhanced to add the + * {@link KafkaHeaders#REPLY_TOPIC} unless it already contains such a header. The handler + * will not map any additional headers; providing such a payload assumes the headers have + * already been mapped. * * @param the key type. * @param the value type. @@ -137,6 +146,8 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes (message, topic, partition, timestamp, key, value, headers) -> new ProducerRecord<>(topic, partition, timestamp, key, value, headers); + private volatile byte[] singleReplyTopic; + public KafkaProducerMessageHandler(final KafkaTemplate kafkaTemplate) { Assert.notNull(kafkaTemplate, "kafkaTemplate cannot be null"); this.kafkaTemplate = kafkaTemplate; @@ -373,6 +384,47 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes @SuppressWarnings("unchecked") @Override protected Object handleRequestMessage(final Message message) { + final ProducerRecord producerRecord; + boolean preBuilt = message.getPayload() instanceof ProducerRecord; + if (preBuilt) { + producerRecord = (ProducerRecord) message.getPayload(); + } + else { + producerRecord = createProducerRecord(message); + } + ListenableFuture> sendFuture; + RequestReplyFuture gatewayFuture = null; + if (this.isGateway && (!preBuilt || producerRecord.headers().lastHeader(KafkaHeaders.REPLY_TOPIC) == null)) { + producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message))); + gatewayFuture = ((ReplyingKafkaTemplate) this.kafkaTemplate).sendAndReceive(producerRecord); + sendFuture = gatewayFuture.getSendFuture(); + } + else { + if (this.transactional + && TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null) { + sendFuture = this.kafkaTemplate.executeInTransaction(t -> { + return t.send(producerRecord); + }); + } + else { + sendFuture = this.kafkaTemplate.send(producerRecord); + } + } + try { + processSendResult(message, producerRecord, sendFuture, getSendSuccessChannel()); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessageHandlingException(message, e); + } + catch (ExecutionException e) { + throw new MessageHandlingException(message, e.getCause()); + } + return processReplyFuture(gatewayFuture); + } + + @SuppressWarnings("unchecked") + private ProducerRecord createProducerRecord(final Message message) { MessageHeaders messageHeaders = message.getHeaders(); String topic = this.topicExpression != null ? this.topicExpression.getValue(this.evaluationContext, message, String.class) @@ -407,35 +459,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } final ProducerRecord producerRecord = this.producerRecordCreator.create(message, topic, partitionId, timestamp, (K) messageKey, payload, headers); - ListenableFuture> sendFuture; - RequestReplyFuture gatewayFuture = null; - if (this.isGateway) { - producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, getReplyTopic(message))); - gatewayFuture = ((ReplyingKafkaTemplate) this.kafkaTemplate).sendAndReceive(producerRecord); - sendFuture = gatewayFuture.getSendFuture(); - } - else { - if (this.transactional - && TransactionSynchronizationManager.getResource(this.kafkaTemplate.getProducerFactory()) == null) { - sendFuture = this.kafkaTemplate.executeInTransaction(t -> { - return t.send(producerRecord); - }); - } - else { - sendFuture = this.kafkaTemplate.send(producerRecord); - } - } - try { - processSendResult(message, producerRecord, sendFuture, getSendSuccessChannel()); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - throw new MessageHandlingException(message, e); - } - catch (ExecutionException e) { - throw new MessageHandlingException(message, e.getCause()); - } - return processReplyFuture(gatewayFuture); + return producerRecord; } private byte[] getReplyTopic(final Message message) { @@ -457,7 +481,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } if (replyTopic == null) { if (this.replyTopicsAndPartitions.size() == 1) { - replyTopic = this.replyTopicsAndPartitions.keySet().iterator().next().getBytes(StandardCharsets.UTF_8); + replyTopic = getSingleReplyTopic(); } else { throw new IllegalStateException("No reply topic header and no default reply topic is can be determined"); @@ -487,6 +511,16 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes return replyTopic; } + private byte[] getSingleReplyTopic() { + if (this.singleReplyTopic == null) { + this.singleReplyTopic = this.replyTopicsAndPartitions.keySet() + .iterator() + .next() + .getBytes(StandardCharsets.UTF_8); + } + return this.singleReplyTopic; + } + private void determineValidReplyTopicsAndPartitions() { ReplyingKafkaTemplate rkt = (ReplyingKafkaTemplate) kafkaTemplate; Collection replyTopics = rkt.getAssignedReplyTopicPartitions(); @@ -544,7 +578,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes } } - private Future processReplyFuture(RequestReplyFuture future) { + private Future processReplyFuture(@Nullable RequestReplyFuture future) { if (future == null) { return null; } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java index 8a7ba0c883..d94932c6ec 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java @@ -352,6 +352,15 @@ class KafkaProducerMessageHandlerTests { @Test void testOutboundGateway() throws Exception { + testOutboundGatewayGuts(null); + } + + @Test + void testOutboundGatewayPrPayload() throws Exception { + testOutboundGatewayGuts(new ProducerRecord(topic5, 1, 2, "foo")); + } + + private void testOutboundGatewayGuts(ProducerRecord payload) throws Exception { ConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory<>( KafkaTestUtils.consumerProps(topic5, "false", embeddedKafka)); ContainerProperties containerProperties = new ContainerProperties(topic6); @@ -384,11 +393,17 @@ class KafkaProducerMessageHandlerTests { handler.setOutputChannel(replies); handler.afterPropertiesSet(); - Message message = MessageBuilder.withPayload("foo") - .setHeader(KafkaHeaders.TOPIC, topic5) - .setHeader(KafkaHeaders.MESSAGE_KEY, 2) - .setHeader(KafkaHeaders.PARTITION_ID, 1) - .build(); + Message message; + if (payload == null) { + message = MessageBuilder.withPayload("foo") + .setHeader(KafkaHeaders.TOPIC, topic5) + .setHeader(KafkaHeaders.MESSAGE_KEY, 2) + .setHeader(KafkaHeaders.PARTITION_ID, 1) + .build(); + } + else { + message = MessageBuilder.withPayload(payload).build(); + } handler.handleMessage(message); ConsumerRecord record = KafkaTestUtils.getSingleRecord(consumer, topic5);