From b6b649396b876e9bea1917f81299f56a85fb4d84 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 18 Jul 2022 13:05:42 -0400 Subject: [PATCH] Fix `KafkaOutAdapterParserTests` race condition The `Future` for `Producer.send()` can be fulfilled before we return it (`isDone()`), so `KafkaTemplate` short-circuits with a `KafkaException`. * Modify `KafkaOutboundAdapterParserTests.testSyncMode()` to assert against `KafkaException` which may be thrown earlier then previously expected `KafkaProducerException` **Cherry-pick to `5.5.x`** --- .../kafka/config/xml/KafkaOutboundAdapterParserTests.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 7e097c0635..a4d2b9e68c 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 @@ -40,7 +40,7 @@ import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.test.util.TestUtils; -import org.springframework.kafka.core.KafkaProducerException; +import org.springframework.kafka.KafkaException; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.core.ProducerFactory; import org.springframework.messaging.MessageHandlingException; @@ -136,7 +136,7 @@ class KafkaOutboundAdapterParserTests { assertThatExceptionOfType(MessageHandlingException.class) .isThrownBy(() -> handler.handleMessage(new GenericMessage<>("foo"))) - .withCauseInstanceOf(KafkaProducerException.class) + .withCauseInstanceOf(KafkaException.class) .withRootCauseInstanceOf(RuntimeException.class) .withMessageContaining("Async Producer Mock exception");