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 123cab2aa5..36dceb7d15 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 @@ -30,6 +30,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.header.Headers; @@ -101,7 +102,10 @@ import org.springframework.util.concurrent.SettableListenableFuture; public class KafkaProducerMessageHandler extends AbstractReplyProducingMessageHandler implements Lifecycle { - private static final long DEFAULT_SEND_TIMEOUT = 10000; + /** + * Buffer added to ensure our timeout is longer than Kafka's. + */ + private static final int TIMEOUT_BUFFER = 5000; private final Map> replyTopicsAndPartitions = new HashMap<>(); @@ -115,6 +119,8 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes private final AtomicBoolean running = new AtomicBoolean(); + private final long deliveryTimeoutMsProperty; + private EvaluationContext evaluationContext; private Expression topicExpression; @@ -130,7 +136,7 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes private boolean sync; - private Expression sendTimeoutExpression = new ValueExpression<>(DEFAULT_SEND_TIMEOUT); + private Expression sendTimeoutExpression; private KafkaHeaderMapper headerMapper; @@ -175,6 +181,25 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes logger.warn("The KafkaTemplate is transactional; this gateway will only work if the consumer is " + "configured to read uncommitted records"); } + determineSendTimeout(); + this.deliveryTimeoutMsProperty = this.sendTimeoutExpression.getValue(Long.class) - TIMEOUT_BUFFER; + } + + private void determineSendTimeout() { + Map props = this.kafkaTemplate.getProducerFactory().getConfigurationProperties(); + Object dt = props.get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG); + if (dt == null) { + dt = ProducerConfig.configDef().defaultValues().get(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG); + } + if (dt instanceof Long) { + setSendTimeout(((Long) dt) + TIMEOUT_BUFFER); + } + else if (dt instanceof Integer) { + setSendTimeout(Long.valueOf((Integer) dt) + TIMEOUT_BUFFER); + } + else if (dt instanceof String) { + setSendTimeout(Long.parseLong((String) dt) + TIMEOUT_BUFFER); + } } public void setTopicExpression(Expression topicExpression) { @@ -243,24 +268,25 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes /** * Specify a timeout in milliseconds for how long this - * {@link KafkaProducerMessageHandler} should wait wait for send operation - * results. Defaults to 10 seconds. The timeout is applied only in {@link #sync} mode. - * Also applies when sending to the success or failure channels. - * @param sendTimeout the timeout to wait for result fo send operation. + * {@link KafkaProducerMessageHandler} should wait wait for send operation results. + * Defaults to the kafka {@code delivery.timeout.ms} property + 5 seconds. The timeout + * is applied Also applies when sending to the success or failure channels. + * @param sendTimeout the timeout to wait for result for a send operation. * @since 2.0.1 */ @Override - public void setSendTimeout(long sendTimeout) { + public final void setSendTimeout(long sendTimeout) { super.setSendTimeout(sendTimeout); setSendTimeoutExpression(new ValueExpression<>(sendTimeout)); } /** * Specify a SpEL expression to evaluate a timeout in milliseconds for how long this - * {@link KafkaProducerMessageHandler} should wait wait for send operation - * results. Defaults to 10 seconds. The timeout is applied only in {@link #sync} mode. + * {@link KafkaProducerMessageHandler} should wait wait for send operation results. + * Defaults to the kafka {@code delivery.timeout.ms} property + 5 seconds. The timeout + * is applied only in {@link #sync} mode. * @param sendTimeoutExpression the {@link Expression} for timeout to wait for result - * fo send operation. + * for a send operation. * @since 2.1.1 */ public void setSendTimeoutExpression(Expression sendTimeoutExpression) { @@ -591,8 +617,15 @@ public class KafkaProducerMessageHandler extends AbstractReplyProducingMes }); } - if (this.sync) { + if (this.sync || this.isGateway) { Long sendTimeout = this.sendTimeoutExpression.getValue(this.evaluationContext, message, Long.class); + if (sendTimeout != null && sendTimeout <= this.deliveryTimeoutMsProperty) { + this.logger.debug("'sendTimeout' increased to " + + (this.deliveryTimeoutMsProperty + TIMEOUT_BUFFER) + + "ms; it must be greater than the 'delivery.timeout.ms' Kafka producer " + + "property to avoid false failures"); + sendTimeout = this.deliveryTimeoutMsProperty + TIMEOUT_BUFFER; + } if (sendTimeout == null || sendTimeout < 0) { future.get(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml index 9dcd1cf8a2..6f6a9ac975 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests-context.xml @@ -38,9 +38,6 @@ - - - + - diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java index 55b00e3472..a2194c84ae 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/config/xml/KafkaOutboundGatewayParserTests.java @@ -17,15 +17,24 @@ package org.springframework.integration.kafka.config.xml; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; + +import java.util.HashMap; +import java.util.Map; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.Bean; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.support.DefaultErrorMessageStrategy; import org.springframework.integration.test.util.TestUtils; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.stereotype.Component; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; @@ -74,8 +83,20 @@ public class KafkaOutboundGatewayParserTests { .isSameAs(this.context.getBean("customHeaderMapper")); } + @Component public static class EMS extends DefaultErrorMessageStrategy { + @SuppressWarnings("rawtypes") + @Bean + public KafkaTemplate template() { + ProducerFactory pf = mock(ProducerFactory.class); + Map props = new HashMap<>(); + given(pf.getConfigurationProperties()).willReturn(props); + KafkaTemplate template = mock(KafkaTemplate.class); + given(template.getProducerFactory()).willReturn(pf); + return template; + } + } } 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 9ae0b49aec..0f2e836cf6 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 @@ -54,6 +54,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.OffsetAndMetadata; import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.Producer; +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.TopicPartition; @@ -147,11 +148,14 @@ class KafkaProducerMessageHandlerTests { @Test void testOutbound() { - DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>( - KafkaTestUtils.producerProps(embeddedKafka)); + Map producerProps = KafkaTestUtils.producerProps(embeddedKafka); + producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 40_000); + DefaultKafkaProducerFactory producerFactory = new DefaultKafkaProducerFactory<>(producerProps); KafkaTemplate template = new KafkaTemplate<>(producerFactory); KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler<>(template); handler.setBeanFactory(mock(BeanFactory.class)); + handler.setSendTimeout(50_000); + handler.setSync(true); handler.afterPropertiesSet(); Message message = MessageBuilder.withPayload("foo") diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index a1022e5c06..a34de03825 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -73,6 +73,10 @@ Flushing after sending several messages might be useful if you are using the `li By default, the expression looks for a `Boolean` value in the `KafkaIntegrationHeaders.FLUSH` header (`kafka_flush`). The flush will occur if the value is `true` and not if it's `false` or the header is absent. +Starting with version 5.4, the `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed from 10 seconds to the `delivery.timeout.ms` Kafka producer property `+ 5000` so that the actual Kafka error after a timeout is propagated to the application, instead of a timeout generated by this framework. +This has been changed for consistency because you may get unexpected behavior (Spring may timeout the send, while it is actually, eventually, successful). +IMPORTANT: That timeout is 120 seconds by default so you may wish to reduce it to get more timely failures. + ==== Java Configuration The following example shows how to configure the Kafka outbound channel adapter with Java: @@ -425,6 +429,10 @@ If your code invokes the gateway behind a synchronous https://docs.spring.io/spr IMPORTANT: The gateway does not accept requests until the reply container has been assigned its topics and partitions. It is suggested that you add a `ConsumerRebalanceListener` to the template's reply container properties and wait for the `onPartitionsAssigned` call before sending messages to the gateway. +Starting with version 5.4, the `KafkaProducerMessageHandler` `sendTimeoutExpression` default has changed from 10 seconds to the `delivery.timeout.ms` Kafka producer property `+ 5000` so that the actual Kafka error after a timeout is propagated to the application, instead of a timeout generated by this framework. +This has been changed for consistency because you may get unexpected behavior (Spring may timeout the send, while it is actually, eventually, successful). +IMPORTANT: That timeout is 120 seconds by default so you may wish to reduce it to get more timely failures. + ==== Java Configuration The following example shows how to configure a gateway with Java: diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index c204bb0323..0ba56c62f3 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -20,6 +20,9 @@ If you are interested in more details, see the Issue Tracker tickets that were r The standalone https://projects.spring.io/spring-integration-kafka/[Spring Integration Kafka] project has been merged as a `spring-integration-kafka` module to this project. 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. + ==== R2DBC Channel Adapters The Channel Adapters for R2DBC database interaction have been introduced.