shashank created CAMEL-26044:
--------------------------------
Summary: camel-dapr - the pub/sub consumer never receives again
after its stream to the Dapr sidecar fails
Key: CAMEL-26044
URL: https://issues.apache.org/jira/browse/CAMEL-26044
Project: Camel
Issue Type: Bug
Components: camel-dapr
Reporter: shashank
Assignee: shashank
The {{dapr:pubSub}} consumer subscribes once, in {{doStart}}, with
{{DaprPreviewClient.subscribeToEvents(pubSubName, topic, listener, type)}}
(main 9f839f4bcc9d). In dapr-sdk 1.18.1 the returned {{Subscription}} runs a
receiver thread that opens the gRPC stream and then waits on a semaphore. Only
the stream observer's {{onCompleted}} releases it, and then a new stream is
opened. The observer's {{onError}} only calls
{{listener.onError(DaprException.propagate(t))}} and returns. So after a failed
stream (the sidecar restarts, {{UNAVAILABLE}}, a reset connection) no new
stream is ever opened. {{DaprPubSubConsumer.DaprSubscriptionListener.onError}}
only logs at ERROR. The route stays started and receives no further event until
it is restarted. The configuration consumer has the same problem (CAMEL-25844).
*Fix*: when the listener gets an error caused by a gRPC
{{StatusRuntimeException}} (the stream failed), the consumer closes the
subscription and subscribes again after a delay: 1 second first, then doubling
up to 30 seconds while subscriptions keep failing. The delay starts again at 1
second once an event is received. The Dapr client also calls {{onError}} for an
event it could not convert and when {{onEvent}} fails, and the stream stays
open then: those errors are only logged, as before. A failed
{{subscribeToEvents}} goes to the consumer's exception handler and is tried
again. One lock guards the subscription and the pending resubscribe; {{doStop}}
cancels a pending resubscribe, and a resubscribe checks {{isRunAllowed()}}. One
WARN per failure; the new subscription request is logged at DEBUG (the Dapr
client opens the stream afterwards), and one INFO on the first event of the new
stream. The component page says that the consumer subscribes again.
*Tests*: new {{DaprPubSubConsumerResubscribeTest}}, 6 tests: a route with a
Mockito {{DaprPreviewClient}} in the registry, and a failed stream simulated as
the Dapr client does it
({{listener.onError(DaprException.propagate(Status.UNAVAILABLE.asRuntimeException()))}}).
* {{subscribesAgainWhenTheStreamFails}} fails on main: {{the consumer must
subscribe again after the stream failed ==> expected: <true> but was: <false>}};
* {{aFailureToSubscribeAgainGoesToTheExceptionHandlerAndIsTriedAgain}},
{{theDelayDoublesWhileTheStreamFailsAndStartsAgainOnAnEvent}} and
{{aLateErrorOfAReplacedSubscriptionIsIgnored}} cover the other paths of the
resubscribe (a failing {{subscribeToEvents}}, the delay 1 s, 2 s, 4 s and its
reset by an event, a late error of a replaced subscription);
* {{anErrorThatDoesNotEndTheStreamKeepsTheSubscription}} and
{{stoppingTheRouteCancelsAPendingResubscribe}} are controls that pass on main
and with the fix.
Unit tests of camel-dapr with the fix: 70 run, 0 failures.
*Not changed*: {{DaprEndpoint.createConsumer}} does not call
{{configureConsumer}}, so the {{exceptionHandler}} and {{bridgeErrorHandler}}
endpoint options do not reach the dapr consumers (follow-up if wanted).
*How it was found*: propagation of our fix pattern "terminal callback of a
watch/stream only logs, no resubscribe" (CAMEL-25730, CAMEL-25680,
CAMEL-25844), with the library behaviour read from the dapr-sdk 1.18.1 jar
({{Subscription}}, {{Subscription$1}}).
*Affected*: main, camel-4.23.x, camel-4.22.x and camel-4.18.x (same consumer).
Read, not run.
_Filed with Claude Code on behalf of allthingssecurity._
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
