[
https://issues.apache.org/jira/browse/CAMEL-25869?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
shashank updated CAMEL-25869:
-----------------------------
Description:
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._
was:
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._
> 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
>
> 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
(v8.20.10#820010)
