diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java index 08aef2d721..cec2e5cdfe 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.java @@ -353,6 +353,28 @@ public class KafkaProducerMessageHandlerSpec extends AbstractReplyProducingMes private String sendSuccessChannelName; + private MessageChannel futuresChannel; + + private String futuresChannelName; + private ErrorMessageStrategy errorMessageStrategy = new DefaultErrorMessageStrategy(); private Type replyPayloadType = Object.class; @@ -338,6 +342,24 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes this.sendSuccessChannelName = sendSuccessChannelName; } + /** + * Set the futures channel. + * @param futuresChannel the futures channel. + * @since 5.4 + */ + public void setFuturesChannel(MessageChannel futuresChannel) { + this.futuresChannel = futuresChannel; + } + + /** + * Set the futures channel name. + * @param futuresChannelName the futures channel name. + * @since 5.4 + */ + public void setFuturesChannelName(String futuresChannelName) { + this.futuresChannelName = futuresChannelName; + } + /** * Set the error message strategy implementation to use when sending error messages after * send failures. Cannot be null. @@ -409,6 +431,17 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes return null; } + protected MessageChannel getFuturesChannel() { + if (this.futuresChannel != null) { + return this.futuresChannel; + } + else if (this.futuresChannelName != null) { + this.futuresChannel = getChannelResolver().resolveDestination(this.futuresChannelName); + return this.futuresChannel; + } + return null; + } + @Override protected void doInit() { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); @@ -447,6 +480,10 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes producerRecord.headers().remove(KafkaIntegrationHeaders.FLUSH); } } + Object futureToken = message.getHeaders().get(KafkaIntegrationHeaders.FUTURE_TOKEN); + if (futureToken != null) { + producerRecord.headers().remove(KafkaIntegrationHeaders.FUTURE_TOKEN); + } ListenableFuture> sendFuture; RequestReplyFuture gatewayFuture = null; if (this.isGateway && (!preBuilt || producerRecord.headers().lastHeader(KafkaHeaders.REPLY_TOPIC) == null)) { @@ -464,6 +501,10 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes sendFuture = this.kafkaTemplate.send(producerRecord); } } + sendFutureIfRequested(message, sendFuture, futureToken); + if (flush) { + this.kafkaTemplate.flush(); + } try { processSendResult(message, producerRecord, sendFuture, getSendSuccessChannel()); } @@ -474,12 +515,28 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes catch (ExecutionException e) { throw new MessageHandlingException(message, e.getCause()); // NOSONAR } - if (flush) { - this.kafkaTemplate.flush(); - } return processReplyFuture(gatewayFuture); } + private void sendFutureIfRequested(final Message message, ListenableFuture> sendFuture, + Object futureToken) { + + if (futureToken != null) { + MessageChannel futures = getFuturesChannel(); + if (futures != null) { + try { + futures.send(getMessageBuilderFactory() + .withPayload(sendFuture) + .setHeader(KafkaIntegrationHeaders.FUTURE_TOKEN, futureToken) + .build()); + } + catch (Exception e) { + this.logger.error(e, "Failed to send sendFuture"); + } + } + } + } + @SuppressWarnings("unchecked") private ProducerRecord createProducerRecord(final Message message) { MessageHeaders messageHeaders = message.getHeaders(); @@ -540,7 +597,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes replyTopic = getSingleReplyTopic(); } else { - throw new IllegalStateException("No reply topic header and no default reply topic is can be determined"); + throw new IllegalStateException("No reply topic header and no default reply topic can be determined"); } } else { diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java index 39bb3ccaaa..7a940d4b79 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaIntegrationHeaders.java @@ -36,4 +36,9 @@ public final class KafkaIntegrationHeaders { */ public static final String FLUSH = KafkaHeaders.PREFIX + "flush"; + /** + * Set to a token to correlate a send Future. + */ + public static final String FUTURE_TOKEN = KafkaHeaders.PREFIX + "futureToken"; + } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java index aaf7ae0c2b..46c8e0dd3c 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundAdapterParserTests.java @@ -22,10 +22,13 @@ import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; import java.time.Duration; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.TimeoutException; import org.apache.kafka.clients.producer.MockProducer; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.common.serialization.IntegerSerializer; import org.apache.kafka.common.serialization.StringSerializer; import org.junit.jupiter.api.Test; @@ -109,12 +112,16 @@ class KafkaOutboundAdapterParserTests { @SuppressWarnings("unchecked") ProducerFactory pf = mock(ProducerFactory.class); given(pf.createProducer()).willReturn(mockProducer); + Map props = new HashMap<>(); + props.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 10); + given(pf.getConfigurationProperties()).willReturn(props); KafkaTemplate template = new KafkaTemplate<>(pf); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); handler.setBeanFactory(mock(BeanFactory.class)); handler.afterPropertiesSet(); handler.setSync(true); + handler.setSendTimeout(10); handler.setTopicExpression(new LiteralExpression("foo")); Executors.newSingleThreadExecutor() diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java index 9ad8aec0ea..965b3cdff8 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java @@ -23,6 +23,7 @@ import java.util.Collection; import java.util.Collections; import java.util.Map; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.stream.Stream; @@ -49,6 +50,7 @@ import org.springframework.integration.kafka.channel.PollableKafkaChannel; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; import org.springframework.integration.kafka.inbound.KafkaMessageSource; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; +import org.springframework.integration.kafka.support.KafkaIntegrationHeaders; import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.test.util.TestUtils; @@ -137,6 +139,9 @@ public class KafkaDslTests { @Autowired private PollableChannel errorChannel; + @Autowired + private PollableChannel futuresChannel; + @Autowired(required = false) @Qualifier("topic1ListenerContainer") private MessageListenerContainer messageListenerContainer; @@ -164,6 +169,12 @@ public class KafkaDslTests { assertThat(TestUtils.getPropertyValue(this.kafkaProducer1, "headerMapper")).isSameAs(this.mapper); + for (int i = 0; i < 200; i++) { + Message future = this.futuresChannel.receive(10000); + assertThat(future).isNotNull(); + ((Future) future.getPayload()).get(10, TimeUnit.SECONDS); + } + for (int i = 0; i < 100; i++) { Message receive = this.listeningFromKafkaResults1.receive(20000); assertThat(receive).isNotNull(); @@ -327,10 +338,16 @@ public class KafkaDslTests { return new DefaultKafkaProducerFactory<>(props); } + @Bean + public PollableChannel futuresChannel() { + return new QueueChannel(); + } + @Bean public IntegrationFlow sendToKafkaFlow() { return f -> f .split(p -> Stream.generate(() -> p).limit(101).iterator(), null) + .enrichHeaders(h -> h.header(KafkaIntegrationHeaders.FUTURE_TOKEN, "foo")) .publishSubscribeChannel(c -> c .subscribe(sf -> sf.handle( kafkaMessageHandler(producerFactory(), TEST_TOPIC1) @@ -354,6 +371,7 @@ public class KafkaDslTests { return Kafka .outboundChannelAdapter(producerFactory) + .futuresChannel("futuresChannel") .sync(true) .messageKey(m -> m .getHeaders() diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 757773da4f..2cc87f8f42 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -847,3 +847,81 @@ public IntegrationFlow flow() { ==== When an integration flow starts with an interface, the proxy that is created has the name of the flow bean, appended with ".gateway" so this bean name can be used a a `@Qualifier` if needed. + +[[read-process-write]] +=== Performance Considerations for read/process/write Scenarios + +Many applications consume from a topic, perform some processing and write to another topic. +In most, cases, if the write fails, the application would want to throw an exception so the incoming request can be retried and/or sent to a dead letter topic. +This functionality is supported by the underlying message listener container, together with a suitably configured error handler. +However, in order to support this, we need to block the listener thread until the success (or failure) of the write operation so that any exceptions can be thrown to the container. +When consuming single records, this is achieved by setting the `sync` property on the outbound adapter. +However, when consuming batches, using `sync` causes a significant performance degradation because the application would wait for the result of each send before sending the next message. +Starting with version 5.4, you can now perform multiple sends and then wait for the results of those sends afterwards. +This is achieved by adding a `futuresChannel` to the message handler. +To enable the feature add `KafkaIntegrationHeaders.FUTURE_TOKEN` to the outbound messages; this can then be used to correlate a `Future` to a particular sent message. +Here is an example of how you might use this feature: + +==== +[source, java] +---- +@SpringBootApplication +public class FuturesChannelApplication { + + public static void main(String[] args) { + SpringApplication.run(FuturesChannelApplication.class, args); + } + + @Bean + IntegrationFlow inbound(ConsumerFactory consumerFactory, Handler handler) { + return IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(consumerFactory, ListenerMode.batch, "inTopic")) + .handle(handler) + .get(); + } + + @Bean + IntegrationFlow outbound(KafkaTemplate kafkaTemplate) { + return IntegrationFlows.from(Gate.class) + .enrichHeaders(h -> h + .header(KafkaHeaders.TOPIC, "outTopic") + .headerExpression(KafkaIntegrationHeaders.FUTURE_TOKEN, "headers[id]")) + .handle(Kafka.outboundChannelAdapter(kafkaTemplate) + .futuresChannel("futures")) + .get(); + } + + @Bean + PollableChannel futures() { + return new QueueChannel(); + } + +} + +@Component +@DependsOn("outbound") +class Handler { + + @Autowired + Gate gate; + + @Autowired + PollableChannel futures; + + public void handle(List input) throws Exception { + System.out.println(input); + input.forEach(str -> this.gate.send(str.toUpperCase())); + for (int i = 0; i < input.size(); i++) { + Message future = this.futures.receive(10000); + ((Future) future.getPayload()).get(10, TimeUnit.SECONDS); + } + } + +} + +interface Gate { + + void send(String out); + +} +---- +==== diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 605104b36a..98d8690385 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -23,6 +23,9 @@ See <<./kafka.adoc#kafka,Spring for Apache Kafka Support>> for more information. The `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed. See <<./kafka.adoc#kafka-outbound,Kafka Outbound Channel Adapter>> for more information. +You can now access the `Future` for underlying `send()` operations. +See <<./kafka.adoc#read-process-write>> for more information. + [[x5.4-r2dbc]] ==== R2DBC Channel Adapters