This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new b62e90ddb2d9 CAMEL-25869: camel-pulsar - a consumer that cannot
subscribe fails the start of the route (#28385)
b62e90ddb2d9 is described below
commit b62e90ddb2d91fd655833191431b8978771c4908
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 24 19:21:06 2026 +0530
CAMEL-25869: camel-pulsar - a consumer that cannot subscribe fails the
start of the route (#28385)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../org/apache/camel/catalog/docs/test-dst.adoc | 4 +-
components/camel-pulsar/pom.xml | 5 +
.../consumers/CommonCreationStrategyImpl.java | 43 ++
.../utils/consumers/FailoverConsumerStrategy.java | 30 +-
.../utils/consumers/KeySharedConsumerStrategy.java | 30 +-
.../utils/consumers/SharedConsumerStrategy.java | 30 +-
.../pulsar/PulsarConsumerSubscribeFailureTest.java | 99 ++++
.../camel/component/pulsar/dst/FakePulsar.java | 525 +++++++++++++++++++++
.../pulsar/dst/PulsarConsumerDstTest.java | 49 ++
.../camel/component/pulsar/dst/PulsarWorld.java | 254 ++++++++++
.../src/test/resources/log4j2.properties | 10 +-
.../camel-test-dst/src/main/docs/test-dst.adoc | 4 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 12 +
13 files changed, 1007 insertions(+), 88 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
index 74a55ce94812..0d13aa8d264a 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/test-dst.adoc
@@ -48,13 +48,15 @@ A `Dst` is one run. The extension creates one per seed and
closes it after the s
`new Dst` while another run is open throws `IllegalStateException` ("another
run (seed N) is still active"), and a DST
test refuses to start when JUnit runs tests in parallel
(`junit.jupiter.execution.parallel.enabled=true`).
-Three worlds are in the Camel code base and are the examples to copy:
+These worlds are in the Camel code base and are the examples to copy:
|===
|Component |Test |Regression test for
|camel-aws2-sqs |`SqsVisibilityDstTest` (the visibility timeout extender)
|CAMEL-25700
|camel-azure-eventhubs |`EventHubsCheckpointDstTest` (checkpointing by batch
size and timeout) |CAMEL-25683
+|camel-pulsar |`PulsarConsumerDstTest` (subscribe failures, broker and route
restarts while the route starts and
+consumes) |CAMEL-25869
|camel-slack |`SlackSuspendDstTest` (route suspend in the middle of a poll)
|CAMEL-25684
|===
diff --git a/components/camel-pulsar/pom.xml b/components/camel-pulsar/pom.xml
index 5d7f37e0aefa..1e8d1347384d 100644
--- a/components/camel-pulsar/pom.xml
+++ b/components/camel-pulsar/pom.xml
@@ -70,6 +70,11 @@
<version>${mockito-version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-dst</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter</artifactId>
diff --git
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/CommonCreationStrategyImpl.java
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/CommonCreationStrategyImpl.java
index fc7d33556839..3e5922a0ec5a 100644
---
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/CommonCreationStrategyImpl.java
+++
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/CommonCreationStrategyImpl.java
@@ -16,6 +16,9 @@
*/
package org.apache.camel.component.pulsar.utils.consumers;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import org.apache.camel.component.pulsar.PulsarConfiguration;
@@ -23,16 +26,56 @@ import org.apache.camel.component.pulsar.PulsarConsumer;
import org.apache.camel.component.pulsar.PulsarEndpoint;
import org.apache.camel.component.pulsar.PulsarMessageListener;
import org.apache.camel.util.ObjectHelper;
+import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.DeadLetterPolicy.DeadLetterPolicyBuilder;
import org.apache.pulsar.client.api.KeySharedPolicy;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public final class CommonCreationStrategyImpl {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(CommonCreationStrategyImpl.class);
+
private CommonCreationStrategyImpl() {
}
+ /**
+ * Subscribes the configured number of consumers.
+ *
+ * When one of them cannot subscribe, the consumers subscribed so far are
closed and the failure is thrown. The
+ * route then fails to start, so it can be started again (for example by
the supervising route controller), instead
+ * of running with fewer consumers than configured, or with none, and
never subscribing the missing ones.
+ */
+ static Collection<Consumer<byte[]>> createMultipleConsumers(
+ final PulsarEndpoint pulsarEndpoint, final PulsarConsumer
pulsarConsumer, final SubscriptionType subscriptionType)
+ throws PulsarClientException {
+ final PulsarConfiguration configuration =
pulsarEndpoint.getPulsarConfiguration();
+ final List<Consumer<byte[]>> consumers = new ArrayList<>();
+
+ try {
+ for (int i = 0; i < configuration.getNumberOfConsumers(); i++) {
+ final String consumerName =
configuration.getConsumerNamePrefix() + i;
+ consumers.add(getBuilder(consumerName, pulsarEndpoint,
pulsarConsumer).subscriptionType(subscriptionType)
+ .subscribe());
+ }
+ } catch (PulsarClientException | RuntimeException e) {
+ for (Consumer<byte[]> consumer : consumers) {
+ try {
+ consumer.close();
+ } catch (Exception closeFailure) {
+ LOGGER.debug("Error closing consumer {} after a failed
subscribe. This exception is ignored.",
+ consumer, closeFailure);
+ }
+ }
+ throw e;
+ }
+ return consumers;
+ }
+
protected static ConsumerBuilder<byte[]> getBuilder(
final String name, final PulsarEndpoint pulsarEndpoint, final
PulsarConsumer pulsarConsumer) {
final PulsarConfiguration endpointConfiguration =
pulsarEndpoint.getPulsarConfiguration();
diff --git
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/FailoverConsumerStrategy.java
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/FailoverConsumerStrategy.java
index 2758015715eb..8e5982ed4d09 100644
---
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/FailoverConsumerStrategy.java
+++
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/FailoverConsumerStrategy.java
@@ -17,22 +17,15 @@
package org.apache.camel.component.pulsar.utils.consumers;
import java.util.Collection;
-import java.util.LinkedList;
-import org.apache.camel.component.pulsar.PulsarConfiguration;
import org.apache.camel.component.pulsar.PulsarConsumer;
import org.apache.camel.component.pulsar.PulsarEndpoint;
import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class FailoverConsumerStrategy implements ConsumerCreationStrategy {
- private static final Logger LOGGER =
LoggerFactory.getLogger(FailoverConsumerStrategy.class);
-
private final PulsarConsumer pulsarConsumer;
FailoverConsumerStrategy(PulsarConsumer pulsarConsumer) {
@@ -40,26 +33,7 @@ public class FailoverConsumerStrategy implements
ConsumerCreationStrategy {
}
@Override
- public Collection<Consumer<byte[]>> create(final PulsarEndpoint
pulsarEndpoint) {
- return createMultipleConsumers(pulsarEndpoint);
- }
-
- private Collection<Consumer<byte[]>> createMultipleConsumers(final
PulsarEndpoint pulsarEndpoint) {
- final Collection<Consumer<byte[]>> consumers = new LinkedList<>();
- final PulsarConfiguration configuration =
pulsarEndpoint.getPulsarConfiguration();
-
- for (int i = 0; i < configuration.getNumberOfConsumers(); i++) {
- final String consumerName = configuration.getConsumerNamePrefix()
+ i;
- try {
- ConsumerBuilder<byte[]> builder
- = CommonCreationStrategyImpl.getBuilder(consumerName,
pulsarEndpoint, pulsarConsumer);
-
-
consumers.add(builder.subscriptionType(SubscriptionType.Failover).subscribe());
- } catch (PulsarClientException exception) {
- LOGGER.error("A PulsarClientException occurred when creating
Consumer {}, {}", consumerName,
- exception.getMessage(), exception);
- }
- }
- return consumers;
+ public Collection<Consumer<byte[]>> create(final PulsarEndpoint
pulsarEndpoint) throws PulsarClientException {
+ return
CommonCreationStrategyImpl.createMultipleConsumers(pulsarEndpoint,
pulsarConsumer, SubscriptionType.Failover);
}
}
diff --git
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/KeySharedConsumerStrategy.java
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/KeySharedConsumerStrategy.java
index 5937b5dc4ecb..77ab71e1258a 100644
---
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/KeySharedConsumerStrategy.java
+++
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/KeySharedConsumerStrategy.java
@@ -17,22 +17,15 @@
package org.apache.camel.component.pulsar.utils.consumers;
import java.util.Collection;
-import java.util.LinkedList;
-import org.apache.camel.component.pulsar.PulsarConfiguration;
import org.apache.camel.component.pulsar.PulsarConsumer;
import org.apache.camel.component.pulsar.PulsarEndpoint;
import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class KeySharedConsumerStrategy implements ConsumerCreationStrategy {
- private static final Logger LOGGER =
LoggerFactory.getLogger(KeySharedConsumerStrategy.class);
-
private final PulsarConsumer pulsarConsumer;
KeySharedConsumerStrategy(PulsarConsumer pulsarConsumer) {
@@ -40,26 +33,7 @@ public class KeySharedConsumerStrategy implements
ConsumerCreationStrategy {
}
@Override
- public Collection<Consumer<byte[]>> create(final PulsarEndpoint
pulsarEndpoint) {
- return createMultipleConsumers(pulsarEndpoint);
- }
-
- private Collection<Consumer<byte[]>> createMultipleConsumers(final
PulsarEndpoint pulsarEndpoint) {
- final Collection<Consumer<byte[]>> consumers = new LinkedList<>();
- final PulsarConfiguration configuration =
pulsarEndpoint.getPulsarConfiguration();
-
- for (int i = 0; i < configuration.getNumberOfConsumers(); i++) {
- final String consumerName = configuration.getConsumerNamePrefix()
+ i;
- try {
- ConsumerBuilder<byte[]> builder
- = CommonCreationStrategyImpl.getBuilder(consumerName,
pulsarEndpoint, pulsarConsumer);
-
-
consumers.add(builder.subscriptionType(SubscriptionType.Key_Shared).subscribe());
- } catch (PulsarClientException exception) {
- LOGGER.error("A PulsarClientException occurred when creating
Consumer {}, {}", consumerName,
- exception.getMessage(), exception);
- }
- }
- return consumers;
+ public Collection<Consumer<byte[]>> create(final PulsarEndpoint
pulsarEndpoint) throws PulsarClientException {
+ return
CommonCreationStrategyImpl.createMultipleConsumers(pulsarEndpoint,
pulsarConsumer, SubscriptionType.Key_Shared);
}
}
diff --git
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/SharedConsumerStrategy.java
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/SharedConsumerStrategy.java
index cf422e90f70e..a2d0210bf910 100644
---
a/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/SharedConsumerStrategy.java
+++
b/components/camel-pulsar/src/main/java/org/apache/camel/component/pulsar/utils/consumers/SharedConsumerStrategy.java
@@ -17,22 +17,15 @@
package org.apache.camel.component.pulsar.utils.consumers;
import java.util.Collection;
-import java.util.LinkedList;
-import org.apache.camel.component.pulsar.PulsarConfiguration;
import org.apache.camel.component.pulsar.PulsarConsumer;
import org.apache.camel.component.pulsar.PulsarEndpoint;
import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class SharedConsumerStrategy implements ConsumerCreationStrategy {
- private static final Logger LOGGER =
LoggerFactory.getLogger(SharedConsumerStrategy.class);
-
private final PulsarConsumer pulsarConsumer;
SharedConsumerStrategy(PulsarConsumer pulsarConsumer) {
@@ -40,26 +33,7 @@ public class SharedConsumerStrategy implements
ConsumerCreationStrategy {
}
@Override
- public Collection<Consumer<byte[]>> create(final PulsarEndpoint
pulsarEndpoint) {
- return createMultipleConsumers(pulsarEndpoint);
- }
-
- private Collection<Consumer<byte[]>> createMultipleConsumers(final
PulsarEndpoint pulsarEndpoint) {
- final Collection<Consumer<byte[]>> consumers = new LinkedList<>();
- final PulsarConfiguration configuration =
pulsarEndpoint.getPulsarConfiguration();
-
- for (int i = 0; i < configuration.getNumberOfConsumers(); i++) {
- final String consumerName = configuration.getConsumerNamePrefix()
+ i;
- try {
- ConsumerBuilder<byte[]> builder
- = CommonCreationStrategyImpl.getBuilder(consumerName,
pulsarEndpoint, pulsarConsumer);
-
-
consumers.add(builder.subscriptionType(SubscriptionType.Shared).subscribe());
- } catch (PulsarClientException exception) {
- LOGGER.error("A PulsarClientException occurred when creating
Consumer {}, {}", consumerName,
- exception.getMessage(), exception);
- }
- }
- return consumers;
+ public Collection<Consumer<byte[]>> create(final PulsarEndpoint
pulsarEndpoint) throws PulsarClientException {
+ return
CommonCreationStrategyImpl.createMultipleConsumers(pulsarEndpoint,
pulsarConsumer, SubscriptionType.Shared);
}
}
diff --git
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarConsumerSubscribeFailureTest.java
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarConsumerSubscribeFailureTest.java
new file mode 100644
index 000000000000..ff038baded58
--- /dev/null
+++
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/PulsarConsumerSubscribeFailureTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.pulsar;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Answers.RETURNS_SELF;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * With several consumers (Shared, Failover, Key_Shared), a consumer whose
subscribe failed was only logged: the route
+ * reported Started with fewer consumers than configured, or with none at all,
and nothing ever subscribed again.
+ */
+class PulsarConsumerSubscribeFailureTest {
+
+ private static final String ROUTE_ID = "pulsar-subscribe-failure";
+
+ @ParameterizedTest
+ @ValueSource(strings = { "SHARED", "FAILOVER", "KEY_SHARED" })
+ void aFailedSubscribeFailsTheRouteStartAndARetryOpensAllConsumers(String
subscriptionType) throws Exception {
+ final List<Consumer<byte[]>> subscribed = new ArrayList<>();
+ final AtomicInteger subscribeCalls = new AtomicInteger();
+
+ final ConsumerBuilder<byte[]> builder = mock(ConsumerBuilder.class,
RETURNS_SELF);
+ when(builder.subscribe()).thenAnswer(invocation -> {
+ // the second consumer of the first start cannot subscribe, e.g.
the broker is unreachable for longer
+ // than the operation timeout; every later subscribe succeeds
+ if (subscribeCalls.incrementAndGet() == 2) {
+ throw new PulsarClientException.TimeoutException("simulated
subscribe timeout");
+ }
+ Consumer<byte[]> consumer = mock(Consumer.class);
+ subscribed.add(consumer);
+ return consumer;
+ });
+ final PulsarClient pulsarClient = mock(PulsarClient.class);
+ when(pulsarClient.newConsumer()).thenReturn(builder);
+
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ final PulsarComponent component = new PulsarComponent(context);
+ component.setPulsarClient(pulsarClient);
+ context.addComponent("pulsar", component);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("pulsar:persistent://public/default/camel-subscribe-failure?subscriptionName=camel-subscription"
+ + "&subscriptionType=" + subscriptionType +
"&numberOfConsumers=2")
+ .routeId(ROUTE_ID).autoStartup(false)
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ // the route must not report Started while it holds only one of
its two consumers: the failure has to
+ // reach the caller (or the supervising route controller), which
can then start it again
+ assertThrows(Exception.class, () ->
context.getRouteController().startRoute(ROUTE_ID));
+ assertEquals(ServiceStatus.Stopped,
context.getRouteController().getRouteStatus(ROUTE_ID));
+ assertEquals(1, subscribed.size());
+ // the consumer that did subscribe must not be left behind, open
and unowned
+ verify(subscribed.get(0)).close();
+
+ context.getRouteController().startRoute(ROUTE_ID);
+ assertEquals(ServiceStatus.Started,
context.getRouteController().getRouteStatus(ROUTE_ID));
+ assertEquals(3, subscribed.size());
+ verify(subscribed.get(1), never()).close();
+ verify(subscribed.get(2), never()).close();
+ }
+ }
+}
diff --git
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/FakePulsar.java
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/FakePulsar.java
new file mode 100644
index 000000000000..206e2c8b19a3
--- /dev/null
+++
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/FakePulsar.java
@@ -0,0 +1,525 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.pulsar.dst;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.TreeSet;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstExecutor;
+import org.apache.camel.test.dst.DstLedger;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageListener;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.SubscriptionInitialPosition;
+import org.apache.pulsar.client.api.SubscriptionType;
+
+/**
+ * A fake Pulsar broker and client for one non-partitioned topic and one
durable subscription, behind the client
+ * interfaces the Camel consumer uses ({@link PulsarClient#newConsumer()},
{@link ConsumerBuilder}, {@link Consumer},
+ * {@link Message}), as dynamic proxies.
+ * <p>
+ * Rules (Pulsar 4.x documentation,
https://pulsar.apache.org/docs/next/concepts-messaging/ and the client Javadoc):
+ * <ul>
+ * <li>[S] "Exclusive ... only a single consumer is allowed to attach to the
subscription. If multiple consumers
+ * subscribe to a topic using the same subscription, an error occurs"
(ConsumerBusyException). "Failover ... the broker
+ * selects the master consumer ... by consumer name order" (assumed: lowest
name). "Shared ... messages are delivered in
+ * a round-robin distribution across consumers". Key_Shared: keys are not
modelled, so it dispatches like Shared.</li>
+ * <li>[Q] receiverQueueSize: "the consumer prefetches messages up to the
receiver queue size"; pause(): "stop
+ * requesting new messages from the broker ... messages already in the
receiver queue are still delivered" (Camel docs
+ * say the same).</li>
+ * <li>[A] acknowledge(): the message is not delivered again; on a closed
consumer it throws
+ * AlreadyClosedException.</li>
+ * <li>[N] negativeAcknowledge(): "the message will be redelivered after a
fixed timeout"
+ * (negativeAckRedeliveryDelay).</li>
+ * <li>[T] ackTimeout: "messages not acknowledged within the timeout are
redelivered"; 0 disables it.</li>
+ * <li>[C] close(): unacknowledged messages are redelivered to the other
consumers of the subscription.</li>
+ * <li>[R] a broker restart / connection loss: the client reconnects by
itself; the broker redelivers every
+ * unacknowledged message and the client clears its receiver queue (assumed:
the whole receiver queue).</li>
+ * <li>[F] subscribe(): fails with a PulsarClientException when the broker
cannot be reached before the operation
+ * timeout (pulsar-client 4.2.4 ConsumerImpl.connectionFailed completes the
subscribe future exceptionally once the
+ * lookup deadline passed, or at once for a non retriable error).</li>
+ * </ul>
+ * Timing: subscribe blocks for its latency and takes effect at the end;
acknowledgements take effect at once; delivery
+ * to the application is a task of a per-consumer listener executor (one at a
time, as the Pulsar listener thread). Not
+ * modelled: receive() and asynchronous acknowledgements (the world uses the
message listener), partitions, keys,
+ * batching, chunking, dead letter topics, retention, ack grouping.
+ */
+final class FakePulsar {
+
+ final Dst dst;
+ private final DstLedger ledger;
+
+ boolean faults = true;
+ double subscribeFailRate;
+
+ /** Published messages, by sequence. */
+ private final List<String> log = new ArrayList<>();
+ /** Messages of the subscription that may be dispatched (not acknowledged,
not out with a consumer). */
+ private final TreeSet<Integer> available = new TreeSet<>();
+ private final TreeSet<Integer> acked = new TreeSet<>();
+ private final Map<Integer, Integer> redeliveries = new LinkedHashMap<>();
+ private final List<FakeConsumer> consumers = new ArrayList<>();
+ private int created;
+ private int rr;
+ private DstExecutor timers;
+
+ FakePulsar(Dst dst, DstLedger ledger) {
+ this.dst = dst;
+ this.ledger = ledger;
+ }
+
+ //
------------------------------------------------------------------------------------------------------------
+ // the world side
+
+ void publish(String id) {
+ int seq = log.size();
+ log.add(id);
+ // the world uses subscriptionInitialPosition=Earliest: a subscription
created later starts at the first message
+ available.add(seq);
+ ledger.record("published", "t", id);
+ dispatch();
+ }
+
+ /** [R] the broker restarts: every consumer reconnects, everything not
acknowledged is delivered again. */
+ void brokerRestart() {
+ dst.trace("fake: broker restart");
+ for (FakeConsumer c : consumers) {
+ if (!c.closed) {
+ returnAll(c);
+ }
+ }
+ dispatch();
+ }
+
+ boolean allAcked() {
+ return acked.size() == log.size();
+ }
+
+ List<String> notAcked() {
+ List<String> answer = new ArrayList<>();
+ for (int i = 0; i < log.size(); i++) {
+ if (!acked.contains(i)) {
+ answer.add(log.get(i));
+ }
+ }
+ return answer;
+ }
+
+ int openConsumers() {
+ return (int) consumers.stream().filter(c -> !c.closed).count();
+ }
+
+ List<String> openConsumerNames() {
+ return consumers.stream().filter(c -> !c.closed).map(c ->
c.name).toList();
+ }
+
+ PulsarClient client() {
+ return proxy(PulsarClient.class, (p, m, a) -> switch (m.getName()) {
+ case "newConsumer" -> new Builder().proxy;
+ case "close", "shutdown" -> null;
+ case "isClosed" -> false;
+ case "toString" -> "FakePulsarClient";
+ case "hashCode" -> System.identityHashCode(p);
+ case "equals" -> p == a[0];
+ default -> throw new UnsupportedOperationException("fake:
unexpected PulsarClient call " + m.getName());
+ });
+ }
+
+ private DstExecutor timers() {
+ if (timers == null) {
+ timers = dst.executor("pulsar-client-timers", 1);
+ }
+ return timers;
+ }
+
+ //
------------------------------------------------------------------------------------------------------------
+ // broker logic
+
+ private void returnAll(FakeConsumer c) {
+ for (Integer seq : c.prefetched) {
+ backToAvailable(seq);
+ }
+ c.prefetched.clear();
+ for (Integer seq : new ArrayList<>(c.unacked.keySet())) {
+ backToAvailable(seq);
+ }
+ c.unacked.clear();
+ }
+
+ private void backToAvailable(int seq) {
+ if (!acked.contains(seq)) {
+ available.add(seq);
+ redeliveries.merge(seq, 1, Integer::sum);
+ }
+ }
+
+ private List<FakeConsumer> dispatchTargets() {
+ List<FakeConsumer> open = consumers.stream().filter(c ->
!c.closed).toList();
+ if (open.isEmpty()) {
+ return open;
+ }
+ SubscriptionType type = open.get(0).type;
+ if (type == SubscriptionType.Failover) {
+ // [S] the active consumer is the first by name
+ FakeConsumer active = open.stream().min((x, y) ->
x.name.compareTo(y.name)).get();
+ return List.of(active);
+ }
+ return open;
+ }
+
+ void dispatch() {
+ List<FakeConsumer> targets = dispatchTargets();
+ if (targets.isEmpty()) {
+ return;
+ }
+ boolean progress = true;
+ while (!available.isEmpty() && progress) {
+ progress = false;
+ for (int i = 0; i < targets.size() && !available.isEmpty(); i++) {
+ FakeConsumer c = targets.get((rr + i) % targets.size());
+ if (!c.paused && c.prefetched.size() < c.queueSize) {
+ Integer seq = available.pollFirst();
+ c.prefetched.add(seq);
+ progress = true;
+ }
+ }
+ rr++;
+ }
+ for (FakeConsumer c : targets) {
+ c.kick();
+ }
+ }
+
+ //
------------------------------------------------------------------------------------------------------------
+ // client side
+
+ private final class Builder {
+ final ConsumerBuilder<byte[]> proxy;
+ String consumerName = "c";
+ SubscriptionType type = SubscriptionType.Exclusive;
+ int queueSize = 1000;
+ long ackTimeoutMillis;
+ long nackDelayMillis = 60000;
+ MessageListener<byte[]> listener;
+ SubscriptionInitialPosition position =
SubscriptionInitialPosition.Latest;
+
+ @SuppressWarnings("unchecked")
+ Builder() {
+ this.proxy = proxy(ConsumerBuilder.class, (p, m, a) -> {
+ switch (m.getName()) {
+ case "consumerName" -> consumerName = (String) a[0];
+ case "subscriptionType" -> type = (SubscriptionType) a[0];
+ case "receiverQueueSize" -> queueSize = (Integer) a[0];
+ case "ackTimeout" -> ackTimeoutMillis = ((TimeUnit)
a[1]).toMillis((Long) a[0]);
+ case "negativeAckRedeliveryDelay" -> nackDelayMillis =
((TimeUnit) a[1]).toMillis((Long) a[0]);
+ case "messageListener" -> listener =
(MessageListener<byte[]>) a[0];
+ case "subscriptionInitialPosition" -> position =
(SubscriptionInitialPosition) a[0];
+ case "subscribe" -> {
+ return subscribe(this);
+ }
+ case "subscribeAsync", "clone", "loadConf" -> throw new
UnsupportedOperationException(
+ "fake: unexpected ConsumerBuilder call " +
m.getName());
+ default -> {
+ // topic, subscriptionName, subscriptionMode,
acknowledgmentGroupTime, readCompacted, ...
+ }
+ }
+ return p;
+ });
+ }
+ }
+
+ private Consumer<byte[]> subscribe(Builder b) throws PulsarClientException
{
+ // [F] the lookup / subscribe round trip; it takes effect at the end
+ long latency = dst.between(1, 50);
+ boolean fail = faults && dst.chance(subscribeFailRate);
+ if (fail) {
+ latency = dst.between(1000, 30000);
+ }
+ dst.block(latency, "subscribe " + b.consumerName);
+ if (fail) {
+ ledger.record("subscribeFailed", "c", b.consumerName);
+ dst.trace("fake: subscribe " + b.consumerName + " failed");
+ throw Dst.injected(new PulsarClientException.TimeoutException(
+ "dst: subscribe " + b.consumerName + " timed out, the
broker could not be reached"));
+ }
+ if (b.type == SubscriptionType.Exclusive && openConsumers() > 0) {
+ throw new PulsarClientException.ConsumerBusyException("Exclusive
consumer is already connected");
+ }
+ FakeConsumer c = new FakeConsumer(b, created++);
+ consumers.add(c);
+ ledger.record("subscribed", "c", c.name);
+ dst.trace("fake: subscribed " + c.name + " (" + b.type + ", open " +
openConsumers() + ")");
+ dispatch();
+ return c.proxy;
+ }
+
+ private final class FakeConsumer {
+ final String name;
+ final SubscriptionType type;
+ final int queueSize;
+ final long ackTimeoutMillis;
+ final long nackDelayMillis;
+ final MessageListener<byte[]> listener;
+ final DstExecutor listenerExecutor;
+ final Deque<Integer> prefetched = new ArrayDeque<>();
+ final Map<Integer, Long> unacked = new LinkedHashMap<>();
+ final Consumer<byte[]> proxy;
+ boolean paused;
+ boolean closed;
+ boolean delivering;
+
+ @SuppressWarnings("unchecked")
+ FakeConsumer(Builder b, int n) {
+ this.name = b.consumerName + "#" + n;
+ this.type = b.type;
+ this.queueSize = Math.max(1, b.queueSize);
+ this.ackTimeoutMillis = b.ackTimeoutMillis;
+ this.nackDelayMillis = b.nackDelayMillis;
+ this.listener = b.listener;
+ this.listenerExecutor = listener != null ?
dst.executor("pulsar-listener-" + name, 1) : null;
+ this.proxy = proxy(Consumer.class, (p, m, a) -> switch
(m.getName()) {
+ case "acknowledge" -> {
+ acknowledge(a[0]);
+ yield null;
+ }
+ case "negativeAcknowledge" -> {
+ negativeAcknowledge(a[0]);
+ yield null;
+ }
+ case "pause" -> {
+ paused = true;
+ dst.trace("fake: pause " + name);
+ yield null;
+ }
+ case "resume" -> {
+ paused = false;
+ dst.trace("fake: resume " + name);
+ dispatch();
+ yield null;
+ }
+ case "close" -> {
+ close();
+ yield null;
+ }
+ case "closeAsync" -> {
+ close();
+ yield CompletableFuture.completedFuture(null);
+ }
+ case "isConnected" -> !closed;
+ case "getConsumerName" -> name;
+ case "getTopic" -> "persistent://public/default/dst";
+ case "getSubscription" -> "subs";
+ case "toString" -> "FakeConsumer[" + name + "]";
+ case "hashCode" -> System.identityHashCode(p);
+ case "equals" -> p == a[0];
+ default -> throw new UnsupportedOperationException("fake:
unexpected Consumer call " + m.getName());
+ });
+ }
+
+ /** Delivers the next prefetched message to the message listener, one
at a time. */
+ void kick() {
+ if (listener == null || delivering || prefetched.isEmpty() ||
closed) {
+ return;
+ }
+ delivering = true;
+ listenerExecutor.schedule("deliver to " + name, dst.between(0, 5),
() -> {
+ delivering = false;
+ if (closed || prefetched.isEmpty()) {
+ return;
+ }
+ Message<byte[]> msg = toApp(prefetched.pollFirst());
+ try {
+ listener.received(proxy, msg);
+ } catch (RuntimeException e) {
+ dst.trace("fake: listener of " + name + " threw " + e);
+ }
+ dispatch();
+ kick();
+ });
+ }
+
+ private Message<byte[]> toApp(int seq) {
+ unacked.put(seq, dst.now());
+ ledger.record("delivered", "t", log.get(seq));
+ if (ackTimeoutMillis > 0) {
+ long at = dst.now();
+ // [T] redelivered when still not acknowledged after the ack
timeout
+ timers().schedule("ack timeout " + log.get(seq),
ackTimeoutMillis, () -> {
+ Long since = unacked.get(seq);
+ if (!closed && since != null && since == at) {
+ unacked.remove(seq);
+ dst.trace("fake: ack timeout of " + log.get(seq) + "
on " + name);
+ backToAvailable(seq);
+ dispatch();
+ }
+ });
+ }
+ return message(seq);
+ }
+
+ private void acknowledge(Object idOrMessage) throws
PulsarClientException {
+ if (closed) {
+ throw new
PulsarClientException.AlreadyClosedException("Consumer already closed");
+ }
+ int seq = seqOf(idOrMessage);
+ unacked.remove(seq);
+ if (!acked.contains(seq)) {
+ acked.add(seq);
+ // [A] an acknowledged message is not delivered again
+ available.remove(seq);
+ }
+ ledger.record("acked", "t", log.get(seq));
+ dispatch();
+ }
+
+ private void negativeAcknowledge(Object idOrMessage) {
+ int seq = seqOf(idOrMessage);
+ if (closed) {
+ return;
+ }
+ Long since = unacked.get(seq);
+ dst.trace("fake: nack " + log.get(seq) + " on " + name);
+ // [N] redelivered after the negative ack redelivery delay
+ timers().schedule("nack redelivery " + log.get(seq),
nackDelayMillis, () -> {
+ if (!closed && since != null &&
Long.valueOf(since).equals(unacked.get(seq))) {
+ unacked.remove(seq);
+ backToAvailable(seq);
+ dispatch();
+ }
+ });
+ }
+
+ private void close() {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ dst.trace("fake: close " + name);
+ ledger.record("closed", "c", name);
+ // [C] the unacknowledged messages go to the other consumers
+ returnAll(this);
+ if (listenerExecutor != null) {
+ listenerExecutor.shutdownNow();
+ }
+ dispatch();
+ }
+ }
+
+ private static int seqOf(Object idOrMessage) {
+ if (idOrMessage instanceof Message<?> m) {
+ return ((Id) m.getMessageId()).seq;
+ }
+ return ((Id) idOrMessage).seq;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Message<byte[]> message(int seq) {
+ String id = log.get(seq);
+ Id messageId = new Id(seq);
+ int redelivery = redeliveries.getOrDefault(seq, 0);
+ return proxy(Message.class, (p, m, a) -> switch (m.getName()) {
+ case "getValue", "getData" -> id.getBytes(StandardCharsets.UTF_8);
+ case "getMessageId" -> messageId;
+ case "getKey" -> id;
+ case "getKeyBytes" -> id.getBytes(StandardCharsets.UTF_8);
+ case "hasKey" -> true;
+ case "getProducerName" -> "dst-producer";
+ case "getTopicName" -> "persistent://public/default/dst";
+ case "getSequenceId" -> (long) seq;
+ case "getPublishTime", "getEventTime" -> Dst.EPOCH_MILLIS + seq;
+ case "getRedeliveryCount" -> redelivery;
+ case "getProperties" -> Collections.emptyMap();
+ case "getProperty" -> null;
+ case "hasProperty" -> false;
+ case "getReaderSchema" -> Optional.empty();
+ case "toString" -> "Message[" + id + "]";
+ case "hashCode" -> seq;
+ case "equals" -> p == a[0];
+ default -> throw new UnsupportedOperationException("fake:
unexpected Message call " + m.getName());
+ });
+ }
+
+ static final class Id implements MessageId {
+ private static final long serialVersionUID = 1L;
+ final int seq;
+
+ Id(int seq) {
+ this.seq = seq;
+ }
+
+ @Override
+ public byte[] toByteArray() {
+ return Integer.toString(seq).getBytes(StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public int compareTo(MessageId o) {
+ return Integer.compare(seq, ((Id) o).seq);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ return o instanceof Id other && other.seq == seq;
+ }
+
+ @Override
+ public int hashCode() {
+ return seq;
+ }
+
+ @Override
+ public String toString() {
+ return "id-" + seq;
+ }
+ }
+
+ @FunctionalInterface
+ interface Handler {
+ Object invoke(Object proxy, Method method, Object[] args) throws
Throwable;
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <T> T proxy(Class<?> type, Handler handler) {
+ InvocationHandler h = (p, m, a) -> {
+ try {
+ return handler.invoke(p, m, a);
+ } catch (InvocationTargetException e) {
+ throw e.getCause();
+ }
+ };
+ return (T) Proxy.newProxyInstance(FakePulsar.class.getClassLoader(),
new Class<?>[] { type }, h);
+ }
+}
diff --git
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/PulsarConsumerDstTest.java
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/PulsarConsumerDstTest.java
new file mode 100644
index 000000000000..0e94031c612a
--- /dev/null
+++
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/PulsarConsumerDstTest.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.pulsar.dst;
+
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstInstrument;
+import org.apache.camel.test.dst.DstSeed;
+import org.apache.camel.test.dst.DstTest;
+
+/**
+ * Deterministic simulation test of the camel-pulsar consumer against a fake
broker, with subscribe failures, broker
+ * restarts, route restarts, suspend and resume, and slow and failing
processing: a started route holds the configured
+ * number of Pulsar consumers and every published message is routed.
Regression test for CAMEL-25869: before that fix
+ * 871 of 5000 seeds fail. A failing seed is replayed with {@code
-Ddst.seed=<seed>}; {@code -Ddst.seeds=5000} runs more
+ * seeds; {@code -Ddst.pulsar.subscribeFailRates=0} runs without subscribe
failures.
+ */
+@DstInstrument(value = "org.apache.camel.component.pulsar",
+ alsoRewrite = { "org.apache.camel.util.StopWatch",
"org.apache.camel.impl.engine.DefaultShutdownStrategy" })
+class PulsarConsumerDstTest {
+
+ @DstTest(seeds = 1000)
+ void aStartedRouteHoldsItsConsumersAndRoutesEveryMessage(Dst dst) throws
Exception {
+ new PulsarWorld(dst).run();
+ }
+
+ /**
+ * The seed that found CAMEL-25869: Key_Shared with one consumer whose
subscribe times out. Before the fix the route
+ * reported Started with no consumer and routed none of the 11 published
messages.
+ */
+ @DstTest
+ @DstSeed(8)
+ void startedAgainAfterItsOnlyConsumerFailedToSubscribe(Dst dst) throws
Exception {
+ new PulsarWorld(dst).run();
+ }
+}
diff --git
a/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/PulsarWorld.java
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/PulsarWorld.java
new file mode 100644
index 000000000000..f3744f947316
--- /dev/null
+++
b/components/camel-pulsar/src/test/java/org/apache/camel/component/pulsar/dst/PulsarWorld.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.pulsar.dst;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.camel.ServiceStatus;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.pulsar.PulsarComponent;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.test.dst.Dst;
+import org.apache.camel.test.dst.DstCamel;
+import org.apache.camel.test.dst.DstExecutor;
+import org.apache.camel.test.dst.DstLedger;
+import org.apache.camel.test.dst.Invariants;
+
+/**
+ * The world around a camel-pulsar consumer: a fake broker and client ({@link
FakePulsar}), a producer, a route with an
+ * asynchronous downstream step that is sometimes slow and sometimes fails,
and an operator that starts the route (and
+ * starts it again when a start fails, as the supervising route controller
does), restarts it, suspends and resumes it.
+ * Faults: subscribe failures (broker unreachable for longer than the
operation timeout), broker restarts.
+ * <p>
+ * Properties: every published message completes in the route at least once;
the subscription is drained once the faults
+ * cleared; a started route holds the configured number of Pulsar consumers
(no fewer: it would not consume, no more:
+ * they would leak).
+ * <p>
+ * Limits: messageListener=true and numberOfConsumerThreads=1 only (the
receive() loops of messageListener=false block
+ * on the one stack until the run ends); no manual acknowledgement, retry or
dead letter topic options.
+ */
+final class PulsarWorld {
+
+ static final String ROUTE = "pulsar";
+
+ private final Dst dst;
+ private final DstLedger ledger;
+ private final FakePulsar pulsar;
+ private final String subscriptionType;
+ private final int numberOfConsumers;
+ private final boolean messageListener;
+ private final int consumerThreads;
+ private final int queueSize;
+ private final long ackTimeoutMillis;
+ private final long nackDelayMillis;
+ private final List<String> violations = new ArrayList<>();
+ private boolean faults = true;
+ private int sendsLeft;
+ private int opsLeft;
+
+ PulsarWorld(Dst dst) {
+ this.dst = dst;
+ this.ledger = new DstLedger(dst);
+ this.pulsar = new FakePulsar(dst, ledger);
+ this.subscriptionType = dst.pick(List.of("EXCLUSIVE", "SHARED",
"FAILOVER", "KEY_SHARED"));
+ this.numberOfConsumers = "EXCLUSIVE".equals(subscriptionType) ? 1 :
dst.pick(List.of(1, 2, 3));
+ this.messageListener = true;
+ this.consumerThreads = 1;
+ this.queueSize = dst.pick(List.of(1, 5, 10));
+ this.ackTimeoutMillis = dst.pick(List.of(0L, 2000L, 10000L));
+ this.nackDelayMillis = dst.pick(List.of(100L, 1000L, 60000L));
+ this.pulsar.subscribeFailRate =
dst.pick(Arrays.stream(System.getProperty("dst.pulsar.subscribeFailRates",
"0,0.1,0.3")
+ .split(",")).map(String::trim).map(Double::valueOf).toList());
+ dst.trace("world: subscriptionType=" + subscriptionType + "
numberOfConsumers=" + numberOfConsumers
+ + " messageListener=" + messageListener + "
numberOfConsumerThreads=" + consumerThreads
+ + " consumerQueueSize=" + queueSize + " ackTimeoutMillis=" +
ackTimeoutMillis
+ + " nackDelayMillis=" + nackDelayMillis + "
subscribeFailRate=" + pulsar.subscribeFailRate);
+ }
+
+ private String config() {
+ return subscriptionType + " consumers=" + numberOfConsumers + "
listener=" + messageListener
+ + " threads=" + consumerThreads + " queue=" + queueSize + "
ackTimeout=" + ackTimeoutMillis
+ + " nack=" + nackDelayMillis + " subscribeFail=" +
pulsar.subscribeFailRate;
+ }
+
+ void run() throws Exception {
+ DefaultCamelContext context = DstCamel.newContext(dst);
+ PulsarComponent component = new PulsarComponent(context);
+ component.setPulsarClient(pulsar.client());
+ context.addComponent("pulsar", component);
+ DstExecutor downstream = dst.executor("downstream", 4);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("pulsar:persistent://public/default/dst?subscriptionName=subs"
+ + "&subscriptionType=" + subscriptionType +
"&numberOfConsumers=" + numberOfConsumers
+ + "&messageListener=" + messageListener +
"&numberOfConsumerThreads=" + consumerThreads
+ + "&consumerQueueSize=" + queueSize +
"&ackTimeoutMillis=" + ackTimeoutMillis
+ + "&negativeAckRedeliveryDelayMicros=" + (nackDelayMillis
* 1000)
+ + "&subscriptionInitialPosition=EARLIEST")
+ .routeId(ROUTE).autoStartup(false)
+ .process(exchange -> {
+ String id = new
String(exchange.getMessage().getBody(byte[].class), StandardCharsets.UTF_8);
+ exchange.setProperty("dstId", id);
+ ledger.record("routed", "t", id);
+ })
+ .process(DstCamel.asyncStep(dst, downstream, 1, 3000,
exchange -> {
+ String id = exchange.getProperty("dstId",
String.class);
+ if (faults && dst.chance(0.05)) {
+ ledger.record("failed", "t", id);
+ throw Dst.injected(new
IllegalStateException("dst: downstream failed for " + id));
+ }
+ ledger.record("completed", "t", id);
+ }));
+ }
+ });
+ context.start();
+ DstExecutor operator = dst.executor("operator", 1);
+ try {
+ operator.schedule("start route", 0, () -> startRoute(context,
operator));
+ planSends();
+ planOperations(context, operator);
+ dst.runUntil(() -> sendsLeft == 0 && opsLeft == 0, "all sends and
operations");
+ dst.runFor(dst.between(0, 5000));
+ faults = false;
+ pulsar.faults = false;
+ long deadline = dst.now() + 5 * 60_000L + 3 * nackDelayMillis;
+ // a start that failed while the faults were on is retried by the
operator: wait for it as well
+ while ((!pulsar.allAcked() ||
context.getRouteController().getRouteStatus(ROUTE) != ServiceStatus.Started)
+ && dst.now() < deadline) {
+ dst.runFor(500);
+ }
+ ServiceStatus status =
context.getRouteController().getRouteStatus(ROUTE);
+ dst.check(
+ Invariants.of("no unexpected client call or consumer
count", () -> violations.isEmpty()
+ ? null : violations.toString()),
+ Invariants.of("the route is started", () -> status ==
ServiceStatus.Started
+ ? null : "route is " + status + " (" + config() +
")"),
+ Invariants.of("a started route holds the configured number
of Pulsar consumers",
+ () -> pulsar.openConsumers() == numberOfConsumers
+ ? null
+ : "open consumers " +
pulsar.openConsumerNames() + ", configured " + numberOfConsumers
+ + " (" + config() + ")"),
+ Invariants.everyOfferedRoutedAtLeastOnce(ledger,
"published", "completed"),
+ Invariants.of("the subscription is drained once the faults
cleared",
+ () -> pulsar.allAcked()
+ ? null : "not acknowledged: " +
pulsar.notAcked() + " (" + config()
+ + ")"));
+ } finally {
+ // stop the route while the answers of the downstream step can
still arrive: an exchange whose answer
+ // dropPending() drops would keep the shutdown waiting on the one
stack
+ try {
+ context.getRouteController().stopRoute(ROUTE);
+ } catch (Exception e) {
+ dst.trace("world: stop route failed (" + e + ")");
+ }
+ dst.dropPending();
+ context.stop();
+ }
+ }
+
+ /** The operator starts the route; a failed start is retried, as the
supervising route controller does. */
+ private void startRoute(DefaultCamelContext context, DstExecutor operator)
{
+ try {
+ context.getRouteController().startRoute(ROUTE);
+ ledger.record("routeStarted", "r", ROUTE);
+ checkConsumerCount("after start");
+ } catch (Exception e) {
+ ledger.record("routeStartFailed", "r", ROUTE);
+ dst.trace("world: start failed (" + e.getMessage() + "), retry in
2 s");
+ operator.schedule("retry start route", 2000, () ->
startRoute(context, operator));
+ }
+ }
+
+ private void checkConsumerCount(String when) {
+ if (pulsar.openConsumers() > numberOfConsumers) {
+ violations.add("more consumers than configured " + when + ": " +
pulsar.openConsumerNames() + " ("
+ + config() + ")");
+ }
+ }
+
+ private void planSends() {
+ DstExecutor producer = dst.executor("producer", 1);
+ long t = dst.between(0, 500);
+ int n = 0;
+ int bursts = (int) dst.between(1, 4);
+ for (int b = 0; b < bursts; b++) {
+ int size = (int) dst.between(1, 12);
+ for (int i = 0; i < size; i++) {
+ String id = "m" + n++;
+ sendsLeft++;
+ producer.schedule("publish " + id, t, () -> {
+ pulsar.publish(id);
+ sendsLeft--;
+ });
+ t += dst.between(0, 100);
+ }
+ t += dst.between(0, 8000);
+ }
+ }
+
+ private void planOperations(DefaultCamelContext context, DstExecutor
operator) {
+ int ops = (int) dst.between(0, 4);
+ long t = dst.between(500, 3000);
+ for (int i = 0; i < ops; i++) {
+ int kind = dst.nextInt(3);
+ opsLeft++;
+ if (kind == 0) {
+ operator.schedule("restart route", t, () -> {
+ try {
+ if (context.getRouteController().getRouteStatus(ROUTE)
== ServiceStatus.Started) {
+ context.getRouteController().stopRoute(ROUTE);
+ ledger.record("routeStopped", "r", ROUTE);
+ if (pulsar.openConsumers() > 0) {
+ violations.add("consumers left open after the
route stopped: "
+ + pulsar.openConsumerNames() +
" (" + config() + ")");
+ }
+ startRoute(context, operator);
+ }
+ } catch (Exception e) {
+ violations.add("restart failed: " + e);
+ }
+ opsLeft--;
+ });
+ } else if (kind == 1) {
+ operator.schedule("suspend and resume route", t, () -> {
+ try {
+ if (context.getRouteController().getRouteStatus(ROUTE)
== ServiceStatus.Started) {
+ context.getRouteController().suspendRoute(ROUTE);
+ dst.block(dst.between(100, 3000), "route
suspended");
+ context.getRouteController().resumeRoute(ROUTE);
+ }
+ } catch (Exception e) {
+ violations.add("suspend/resume failed: " + e);
+ }
+ opsLeft--;
+ });
+ } else {
+ operator.schedule("broker restart", t, () -> {
+ if (faults) {
+ pulsar.brokerRestart();
+ }
+ opsLeft--;
+ });
+ }
+ t += dst.between(500, 8000);
+ }
+ }
+}
diff --git a/components/camel-pulsar/src/test/resources/log4j2.properties
b/components/camel-pulsar/src/test/resources/log4j2.properties
index dddebe493b52..80cf39b40444 100644
--- a/components/camel-pulsar/src/test/resources/log4j2.properties
+++ b/components/camel-pulsar/src/test/resources/log4j2.properties
@@ -15,14 +15,20 @@
## limitations under the License.
## ---------------------------------------------------------------------------
-appender.file.type = File
+appender.file.type = RollingFile
appender.file.name = file
appender.file.fileName = target/camel-pulsar-test.log
+appender.file.filePattern = target/camel-pulsar-test.%i.log
+appender.file.policies.type = Policies
+appender.file.policies.size.type = SizeBasedTriggeringPolicy
+appender.file.policies.size.size = 50MB
+appender.file.strategy.type = DefaultRolloverStrategy
+appender.file.strategy.max = 1
appender.file.layout.type = PatternLayout
appender.file.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
appender.out.type = Console
appender.out.name = out
appender.out.layout.type = PatternLayout
appender.out.layout.pattern = %d [%-15.15t] %-5p %-30.30c{1} - %m%n
-rootLogger.level = INFO
+rootLogger.level = WARN
rootLogger.appenderRef.file.ref = file
diff --git a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
index 74a55ce94812..0d13aa8d264a 100644
--- a/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
+++ b/components/camel-test/camel-test-dst/src/main/docs/test-dst.adoc
@@ -48,13 +48,15 @@ A `Dst` is one run. The extension creates one per seed and
closes it after the s
`new Dst` while another run is open throws `IllegalStateException` ("another
run (seed N) is still active"), and a DST
test refuses to start when JUnit runs tests in parallel
(`junit.jupiter.execution.parallel.enabled=true`).
-Three worlds are in the Camel code base and are the examples to copy:
+These worlds are in the Camel code base and are the examples to copy:
|===
|Component |Test |Regression test for
|camel-aws2-sqs |`SqsVisibilityDstTest` (the visibility timeout extender)
|CAMEL-25700
|camel-azure-eventhubs |`EventHubsCheckpointDstTest` (checkpointing by batch
size and timeout) |CAMEL-25683
+|camel-pulsar |`PulsarConsumerDstTest` (subscribe failures, broker and route
restarts while the route starts and
+consumes) |CAMEL-25869
|camel-slack |`SlackSuspendDstTest` (route suspend in the middle of a poll)
|CAMEL-25684
|===
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 97d2460e984d..4fef939267fc 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -1528,6 +1528,18 @@ they refer to the same `ConnectionConfiguration`
instance (for example the same
`XmppEndpoint.createConnection()` now creates a new connection on every call,
instead of returning the connection of
the endpoint, and an endpoint no longer disconnects a connection when it stops.
+=== camel-pulsar - a consumer that cannot subscribe fails the start of the
route
+
+With `subscriptionType` `Shared`, `Failover` or `Key_Shared`, a consumer that
could not subscribe when the route
+started (for example a broker that could not be reached within the operation
timeout of the Pulsar client) was only
+logged at `ERROR`, and the route started with fewer consumers than
`numberOfConsumers`, or with none, and never
+subscribed them again. Now the consumers that did subscribe are closed and the
route fails to start with the error of
+the Pulsar client, as it already did with `Exclusive`. When the route starts
with the CamelContext, the CamelContext
+then fails to start, unless the supervising route controller
(`camel.routecontroller.enabled=true`) is used, which
+starts such a route again. With `messageListener=true` (the default) the
consumers that did subscribe may already have
+received messages before they are closed, and messages they received but did
not acknowledge can be delivered again
+once the route has started.
+
=== camel-vertx-http - a header with several values is sent as several headers
A message header whose value is a `java.util.Collection` or an object array
(as the HTTP consumers map a request
