[
https://issues.apache.org/jira/browse/CAMEL-26044?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Claus Ibsen resolved CAMEL-26044.
---------------------------------
Resolution: Fixed
Fixed via https://github.com/apache/camel/pull/28532
> 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
> Priority: Major
> Fix For: 4.24.0
>
>
> 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)
