Skip to content

CAMEL-23880: camel-pulsar - fix flaky integration tests #24380

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
gnodet merged 1 commit into from
Jul 3, 2026
Merged

CAMEL-23880: camel-pulsar - fix flaky integration tests #24380

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 52 additions & 21 deletions ...rg/apache/camel/component/pulsar/integration/PulsarConsumerNegativeAcknowledgementIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String> 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
Expand Down Expand Up @@ -94,15 +118,22 @@ private PulsarClient givenPulsarClient() throws PulsarClientException {

@Test
public void testAMessageIsConsumedMultipleTimesWithNegativeAckBackoff() throws Exception {
to.expectedMessageCount(3);
to.expectedMinimumMessageCount(3);

Producer<String> 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();
}
}
75 changes: 54 additions & 21 deletions ...java/org/apache/camel/component/pulsar/integration/PulsarConsumerNoAcknowledgementIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<String> 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
Expand Down Expand Up @@ -89,27 +116,33 @@ private PulsarClient givenPulsarClient() throws PulsarClientException {
public void testAMessageIsConsumedMultipleTimes() throws Exception {
to.expectedMinimumMessageCount(2);

Producer<String> 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<String> 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();
}
}
45 changes: 29 additions & 16 deletions ...est/java/org/apache/camel/component/pulsar/integration/PulsarConsumerReadCompactedIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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<String> 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
}
}

Expand All @@ -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
Expand Down