From 7be7e46fb6bbdc8d210305ae0b5994c56f3ccc6e Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 2 Jul 2026 11:21:14 +0000 Subject: [PATCH] CAMEL-23880: camel-pulsar - fix flaky integration tests Root cause: multiple test classes shared the same Pulsar topic URI (persistent://public/default/camel-topic) with the same subscription name while running against a singleton Pulsar container. This caused messages and subscription state to leak between tests, producing non-deterministic message counts and spurious CI failures on unrelated PRs. Fixes: - Give each test class a unique topic name prefix and each test method its own topic via a static counter, preventing cross-contamination - Replace @EndpointInject for consumer endpoints with programmatic endpoint creation in @BeforeEach, so each test method gets a fresh endpoint bound to its unique topic - Add proper @AfterEach cleanup (close endpoints, producers, routes) - Change expectedMessageCount() to expectedMinimumMessageCount() for redelivery-based assertions whose exact count is timing-dependent - Replace Guava Uninterruptibles.sleepUninterruptibly with Awaitility in PulsarConsumerReadCompactedIT (compaction status polling) Co-Authored-By: Claude Opus 4.6 Signed-off-by: Guillaume Nodet --- ...lsarConsumerNegativeAcknowledgementIT.java | 73 ++++++++++++------ .../PulsarConsumerNoAcknowledgementIT.java | 75 +++++++++++++------ .../PulsarConsumerReadCompactedIT.java | 45 +++++++---- 3 files changed, 135 insertions(+), 58 deletions(-) diff --git a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerNegativeAcknowledgementIT.java b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerNegativeAcknowledgementIT.java index 703ae9b299393..a6970833d94de 100644 --- a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerNegativeAcknowledgementIT.java +++ b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerNegativeAcknowledgementIT.java @@ -27,39 +27,63 @@ import org.apache.camel.component.pulsar.utils.AutoConfiguration; import org.apache.camel.component.pulsar.utils.message.PulsarMessageHeaders; import org.apache.camel.spi.Registry; +import org.apache.camel.test.infra.common.TestUtils; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.ClientBuilderImpl; import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class PulsarConsumerNegativeAcknowledgementIT extends PulsarITSupport { - private static final String TOPIC_URI = "persistent://public/default/camel-topic-negative-ack"; - private static final String PRODUCER = "camel-producer-1"; + private static final Logger LOGGER = LoggerFactory.getLogger(PulsarConsumerNegativeAcknowledgementIT.class); + private static final String TOPIC_URI_PREFIX = "persistent://public/default/camel-neg-ack-topic-"; + private static int topicId; - @EndpointInject("pulsar:" + TOPIC_URI + "?numberOfConsumers=1&subscriptionType=Exclusive&batchingEnabled=false" - + "&subscriptionName=camel-subscription&consumerQueueSize=1&consumerName=camel-consumer") private Endpoint from; @EndpointInject("mock:result") private MockEndpoint to; - @Override - protected RouteBuilder createRouteBuilder() { - return new RouteBuilder() { - @Override - public void configure() { - // This route will explicitly negative acknowledge the message. - from(from) - .process(exchange -> exchange.getIn() - .getHeader(PulsarMessageHeaders.MESSAGE_RECEIPT, PulsarMessageReceipt.class) - .negativeAcknowledge()) - .to(to); + private Producer producer; + + @BeforeEach + public void setup() throws Exception { + context.removeRoute("myRoute"); + String producerName = this.getClass().getSimpleName() + TestUtils.randomWithRange(1, 100); + + String topicUri = TOPIC_URI_PREFIX + ++topicId; + producer = givenPulsarClient().newProducer(Schema.STRING).producerName(producerName).topic(topicUri).create(); + + from = context.getEndpoint("pulsar:" + topicUri + "?numberOfConsumers=1&subscriptionType=Exclusive" + + "&batchingEnabled=false" + + "&subscriptionName=camel-subscription&consumerQueueSize=1&consumerName=camel-consumer"); + to = context.getEndpoint("mock:result", MockEndpoint.class); + } + + @AfterEach + public void tearDown() { + try { + if (from != null) { + from.close(); + } + if (producer != null) { + producer.close(); } - }; + } catch (Exception e) { + LOGGER.warn("Failed to close resources: {}", e.getMessage(), e); + } + try { + context.removeRoute("myRoute"); + } catch (Exception e) { + // ignore + } } @Override @@ -94,15 +118,22 @@ private PulsarClient givenPulsarClient() throws PulsarClientException { @Test public void testAMessageIsConsumedMultipleTimesWithNegativeAckBackoff() throws Exception { - to.expectedMessageCount(3); + to.expectedMinimumMessageCount(3); - Producer producer - = givenPulsarClient().newProducer(Schema.STRING).producerName(PRODUCER).topic(TOPIC_URI).create(); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + // This route will explicitly negative acknowledge the message. + from(from).routeId("myRoute") + .process(exchange -> exchange.getIn() + .getHeader(PulsarMessageHeaders.MESSAGE_RECEIPT, PulsarMessageReceipt.class) + .negativeAcknowledge()) + .to(to); + } + }); producer.send("Hello World!"); MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, to); - - producer.close(); } } diff --git a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerNoAcknowledgementIT.java b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerNoAcknowledgementIT.java index 9f3f2092abb47..f8e75ea7fed68 100644 --- a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerNoAcknowledgementIT.java +++ b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerNoAcknowledgementIT.java @@ -25,35 +25,62 @@ import org.apache.camel.component.pulsar.PulsarComponent; import org.apache.camel.component.pulsar.utils.AutoConfiguration; import org.apache.camel.spi.Registry; +import org.apache.camel.test.infra.common.TestUtils; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.ClientBuilderImpl; import org.apache.pulsar.client.impl.MultiplierRedeliveryBackoff; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class PulsarConsumerNoAcknowledgementIT extends PulsarITSupport { - private static final String TOPIC_URI = "persistent://public/default/camel-topic"; - private static final String PRODUCER = "camel-producer-1"; + private static final Logger LOGGER = LoggerFactory.getLogger(PulsarConsumerNoAcknowledgementIT.class); + private static final String TOPIC_URI_PREFIX = "persistent://public/default/camel-no-ack-topic-"; + private static int topicId; - @EndpointInject("pulsar:" + TOPIC_URI + "?numberOfConsumers=1&subscriptionType=Exclusive" - + "&subscriptionName=camel-subscription&consumerQueueSize=1&consumerName=camel-consumer") private Endpoint from; @EndpointInject("mock:result") private MockEndpoint to; - @Override - protected RouteBuilder createRouteBuilder() { - return new RouteBuilder() { - @Override - public void configure() { - // Nothing in the route will ack the message. - from(from).to(to); + private Producer producer; + + @BeforeEach + public void setup() throws Exception { + context.removeRoute("myRoute"); + String producerName = this.getClass().getSimpleName() + TestUtils.randomWithRange(1, 100); + + String topicUri = TOPIC_URI_PREFIX + ++topicId; + producer = givenPulsarClient().newProducer(Schema.STRING).producerName(producerName).topic(topicUri).create(); + + from = context.getEndpoint("pulsar:" + topicUri + "?numberOfConsumers=1&subscriptionType=Exclusive" + + "&subscriptionName=camel-subscription&consumerQueueSize=1&consumerName=camel-consumer"); + to = context.getEndpoint("mock:result", MockEndpoint.class); + } + + @AfterEach + public void tearDown() { + try { + if (from != null) { + from.close(); + } + if (producer != null) { + producer.close(); } - }; + } catch (Exception e) { + LOGGER.warn("Failed to close resources: {}", e.getMessage(), e); + } + try { + context.removeRoute("myRoute"); + } catch (Exception e) { + // ignore + } } @Override @@ -89,27 +116,33 @@ private PulsarClient givenPulsarClient() throws PulsarClientException { public void testAMessageIsConsumedMultipleTimes() throws Exception { to.expectedMinimumMessageCount(2); - Producer producer - = givenPulsarClient().newProducer(Schema.STRING).producerName(PRODUCER).topic(TOPIC_URI).create(); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + // Nothing in the route will ack the message. + from(from).routeId("myRoute").to(to); + } + }); producer.send("Hello World!"); MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, to); - - producer.close(); } @Test public void testAMessageIsConsumedMultipleTimesWithAckTimeoutBackoff() throws Exception { - to.expectedMessageCount(3); + to.expectedMinimumMessageCount(3); - Producer producer - = givenPulsarClient().newProducer(Schema.STRING).producerName(PRODUCER).topic(TOPIC_URI).create(); + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + // Nothing in the route will ack the message. + from(from).routeId("myRoute").to(to); + } + }); producer.send("Hello World!"); MockEndpoint.assertIsSatisfied(10, TimeUnit.SECONDS, to); - - producer.close(); } } diff --git a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerReadCompactedIT.java b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerReadCompactedIT.java index 1987d66ec4b42..a7b84b2fae5c5 100644 --- a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerReadCompactedIT.java +++ b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/integration/PulsarConsumerReadCompactedIT.java @@ -18,7 +18,6 @@ import java.util.concurrent.TimeUnit; -import com.google.common.util.concurrent.Uninterruptibles; import org.apache.camel.Endpoint; import org.apache.camel.EndpointInject; import org.apache.camel.builder.RouteBuilder; @@ -43,39 +42,54 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import static org.awaitility.Awaitility.await; + public class PulsarConsumerReadCompactedIT extends PulsarITSupport { private static final Logger LOGGER = LoggerFactory.getLogger(PulsarConsumerReadCompactedIT.class); - private static final String TOPIC_URI = "persistent://public/default/camel-topic"; + private static final String TOPIC_URI_PREFIX = "persistent://public/default/camel-read-compacted-topic-"; + private static int topicId; - @EndpointInject("pulsar:" + TOPIC_URI + "?numberOfConsumers=1&subscriptionType=Exclusive" - + "&subscriptionName=camel-subscription&consumerQueueSize=1&consumerName=camel-consumer" - + "&allowManualAcknowledgement=true" + "&ackTimeoutMillis=1000" - + "&readCompacted=true" - + "&negativeAckRedeliveryDelayMicros=100000") private Endpoint from; @EndpointInject("mock:result") private MockEndpoint to; private Producer producer; + private String topicUri; @BeforeEach public void setup() throws Exception { context.removeRoute("myRoute"); + topicUri = TOPIC_URI_PREFIX + ++topicId; String producerName = this.getClass().getSimpleName() + TestUtils.randomWithRange(1, 100); - producer = givenPulsarClient().newProducer(Schema.STRING).producerName(producerName).topic(TOPIC_URI).create(); + producer = givenPulsarClient().newProducer(Schema.STRING).producerName(producerName).topic(topicUri).create(); + + from = context.getEndpoint("pulsar:" + topicUri + "?numberOfConsumers=1&subscriptionType=Exclusive" + + "&subscriptionName=camel-subscription&consumerQueueSize=1&consumerName=camel-consumer" + + "&allowManualAcknowledgement=true" + "&ackTimeoutMillis=1000" + + "&readCompacted=true" + + "&negativeAckRedeliveryDelayMicros=100000"); + to = context.getEndpoint("mock:result", MockEndpoint.class); } @AfterEach - public void tearDownProducer() { + public void tearDown() { try { + if (from != null) { + from.close(); + } if (producer != null) { producer.close(); } - } catch (PulsarClientException e) { - LOGGER.warn("Failed to close client: {}", e.getMessage(), e); + } catch (Exception e) { + LOGGER.warn("Failed to close resources: {}", e.getMessage(), e); + } + try { + context.removeRoute("myRoute"); + } catch (Exception e) { + // ignore } } @@ -100,11 +114,10 @@ private PulsarAdmin givenPulsarAdmin() throws PulsarClientException { private void triggerCompaction() throws PulsarAdminException, PulsarClientException { final Topics topics = givenPulsarAdmin().topics(); - topics.triggerCompaction(TOPIC_URI); - while (topics.compactionStatus(TOPIC_URI).status.equals(LongRunningProcessStatus.Status.RUNNING)) { - LOGGER.info("Waiting for compaction completeness..."); - Uninterruptibles.sleepUninterruptibly(1, TimeUnit.SECONDS); - } + topics.triggerCompaction(topicUri); + await().atMost(30, TimeUnit.SECONDS) + .pollInterval(1, TimeUnit.SECONDS) + .until(() -> !topics.compactionStatus(topicUri).status.equals(LongRunningProcessStatus.Status.RUNNING)); } @Test