shashank created CAMEL-25869:
--------------------------------
Summary: camel-pulsar - with Shared, Failover or Key_Shared
subscriptions, a consumer that cannot subscribe is dropped: the route starts
with fewer consumers than configured, or none, and never subscribes them again
Key: CAMEL-25869
URL: https://issues.apache.org/jira/browse/CAMEL-25869
Project: Camel
Issue Type: Bug
Components: camel-pulsar
Reporter: shashank
Assignee: shashank
With {{subscriptionType}} {{Shared}}, {{Failover}} or {{Key_Shared}}, the
consumer creation strategies ({{SharedConsumerStrategy}},
{{FailoverConsumerStrategy}}, {{KeySharedConsumerStrategy}}, main 387cab56627b,
lines 51-62 of each) subscribe {{numberOfConsumers}} consumers in a loop and
catch the failure of each subscribe:
{code:java}
for (int i = 0; i < configuration.getNumberOfConsumers(); i++) {
final String consumerName = configuration.getConsumerNamePrefix() + i;
try {
...
consumers.add(builder.subscriptionType(SubscriptionType.Shared).subscribe());
} catch (PulsarClientException exception) {
LOGGER.error("A PulsarClientException occurred when creating Consumer
{}, {}", consumerName,
exception.getMessage(), exception);
}
}
return consumers;
{code}
{{PulsarConsumer.doStart}} keeps whatever came back. So when a subscribe fails,
the route still reports Started, with fewer consumers than configured, or with
none, and nothing subscribes the missing ones again: there is no Pulsar
consumer object left that could reconnect. In the Pulsar client (4.2.4,
{{ConsumerImpl.connectionFailed}}), {{subscribe()}} throws once the broker
could not be reached before the operation timeout (30 s by default), or on an
error it does not retry; the consumer is then cleaned up from the client.
Typical cases: the application starts while the broker restarts or is not
reachable yet, a topic lookup that times out.
Consequences:
* {{numberOfConsumers=1}} (the default): the route is Started and holds no
consumer. Nothing is consumed and the messages stay in the subscription until
the route or the application is restarted. Only one ERROR line in the log.
* {{numberOfConsumers=N}}: the route consumes with fewer consumers than
configured, without notice (less throughput; with {{Key_Shared}} and
{{Failover}} the subscription is spread over fewer consumers than planned).
* The supervising route controller cannot help, as the start did not fail.
With {{Exclusive}} the failure is not caught, and the route fails to start
({{ExclusiveConsumerStrategy}}), so the two paths behave differently for the
same broker error.
*Fix*: the three strategies share one loop
({{CommonCreationStrategyImpl.createMultipleConsumers}}). When a subscribe
fails, the consumers subscribed so far are closed and the failure is thrown, so
the route fails to start as with {{Exclusive}}; the caller or the supervising
route controller can start it again, and a start that succeeds holds all the
configured consumers. Upgrade note in the 4.24 guide, as a route that used to
start with missing consumers now fails to start.
*Tests*: new {{PulsarConsumerSubscribeFailureTest}} (parameterised over
{{SHARED}}, {{FAILOVER}}, {{KEY_SHARED}}; a mocked {{PulsarClient}} whose
second subscribe throws {{PulsarClientException.TimeoutException}},
{{numberOfConsumers=2}}). On main all three fail with {{Expected
java.lang.Exception to be thrown, but nothing was thrown}} (the route starts
with one consumer). With the fix: the start fails, the consumer that did
subscribe is closed, and a second start opens both consumers. camel-pulsar unit
tests with the fix: 33 run, 0 failures.
*How it was found*: deterministic simulation of the camel-pulsar consumer
against a fake Pulsar client (subscribe failures, broker restarts, route
restarts, suspend/resume, slow and failing processing; an operator that starts
the route again when its start fails, as the supervising route controller
does). The invariant "a started route holds the configured number of Pulsar
consumers" failed in 57 of 300 seeds, first reported on seed 9: {{FAILOVER
numberOfConsumers=2}}, the subscribe of the first consumer timed out, the route
reported Started with one consumer. In seed 8 ({{KEY_SHARED
numberOfConsumers=1}}) the route was Started with no consumer, and none of the
11 published messages was routed in more than 5 simulated minutes after the
faults cleared. With the fix, 300 of 300 seeds pass; without subscribe
failures, main passes 300 of 300 (no other invariant failed). The JUnit test
above is the evidence; no run against a real broker.
*Affected*: main, camel-4.23.x, camel-4.22.x and camel-4.18.x (the same catch
in the three strategies; read, not run).
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
