[
https://issues.apache.org/jira/browse/CAMEL-25869?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen updated CAMEL-25869:
--------------------------------
Fix Version/s: 4.24.0
> 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
> Priority: Major
> Fix For: 4.24.0
>
>
> 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. The pull
> request also adds the deterministic simulation test that found the bug
> ({{PulsarConsumerDstTest}}, camel-test-dst, 1000 seeds, seed 8 pinned with
> {{@DstSeed}}; see below). camel-pulsar unit tests with the fix: 1034 run (33,
> plus 1001 of the simulation test), 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 world ({{PulsarConsumerDstTest}}, {{PulsarWorld}},
> {{FakePulsar}} in the {{dst}} package of the camel-pulsar tests) is part of
> the pull request. The invariant "a started route holds the configured number
> of Pulsar consumers" fails on main in 871 of 5000 seeds, all with {{Shared}},
> {{Failover}} or {{Key_Shared}} and subscribe failures on; in 215 of them the
> route is Started with no consumer at all. 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, 5000 of 5000 seeds pass; without subscribe
> failures, main passes 1000 of 1000 (no other invariant failed). The JUnit
> test above shows it without the simulation; 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
(v10.3.26#10030026)
