allthingssecurity opened a new pull request, #28532: URL: https://github.com/apache/camel/pull/28532
# Description [CAMEL-26044](https://issues.apache.org/jira/browse/CAMEL-26044) The `dapr:pubSub` consumer subscribes once, in `doStart`, with `DaprPreviewClient.subscribeToEvents(pubSubName, topic, listener, type)`. In dapr-sdk 1.18.1 the returned `Subscription` runs a receiver thread that opens the gRPC stream, then waits on a semaphore. The semaphore is released only by the stream observer's `onCompleted`, which opens a new stream. 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) the receiver thread waits for ever and no new stream is 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, fixed in CAMEL-25844; this is the pub/sub consumer. Changes: - When the listener gets an error whose cause is 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. One WARN line per failure (stack trace at DEBUG). `subscribeToEvents` only sends the request (the Dapr client opens the stream afterwards, and a sidecar that is still down fails it again through `onError`), so the new request is logged at DEBUG, and the INFO line "Receiving events again ... after the subscription failed" is logged on the first event of the new stream. - The Dapr client also calls `onError` for an event it could not convert and when the listener's `onEvent` fails; the stream stays open then. Those errors are still only logged, as before, and do not close the subscription. - If `subscribeToEvents` itself throws when subscribing again, the error goes to the consumer's exception handler and another attempt is scheduled. Note: `DaprEndpoint.createConsumer` does not call `configureConsumer`, so the `exceptionHandler` and `bridgeErrorHandler` endpoint options do not reach the dapr consumers today and that handler is the default logging one. This is not changed here (follow-up JIRA if wanted). - The subscription, its listener and the pending resubscribe are guarded by one lock. `doStop` cancels a pending resubscribe and closes the subscription under it, and a resubscribe checks `isRunAllowed()` under the lock, so a stopped route does not subscribe again. An error from a subscription that was already replaced or closed is ignored. - The delay uses `Mono.delay` (Reactor's parallel scheduler, already used by the Dapr client), so the consumer creates no thread pool. - The component page (and its catalog copy) says that the consumer subscribes again, and how the delay grows. No option or generated metadata changed. No upgrade guide entry: users have nothing to change. I can add one if wanted. - Backport: camel-4.23.x, 4.22.x and 4.18.x have the same consumer. Tests: - New `DaprPubSubConsumerResubscribeTest`, 6 tests, JUnit assertions as in the module. It runs a route `from("dapr:pubSub?...")` with a Mockito `DaprPreviewClient` in the registry (autowired). Each `subscribeToEvents` call returns a new mock `Subscription` and keeps the listener. A failed stream is 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>`. With the fix the first subscription is closed, the new one routes an event, and a second failure subscribes a third time. - Three tests cover the other paths of the resubscribe, which main does not have: - `aFailureToSubscribeAgainGoesToTheExceptionHandlerAndIsTriedAgain`: the 2nd `subscribeToEvents` throws; the exception reaches the consumer's exception handler (set on the consumer, see the note above) and a 3rd call subscribes. - `theDelayDoublesWhileTheStreamFailsAndStartsAgainOnAnEvent`: 1 s, 2 s, 4 s, then back to 1 s after an event (read through a package-private getter for tests). - `aLateErrorOfAReplacedSubscriptionIsIgnored`: an error of the first listener after the consumer subscribed again neither closes the new subscription nor subscribes a third time. - With the stale-listener check removed from `onStreamFailed` and the delay reset removed from `onEvent`, the last two fail (mutation check). - `anErrorThatDoesNotEndTheStreamKeepsTheSubscription` and `stoppingTheRouteCancelsAPendingResubscribe` are controls that pass on main and with the fix. They check that an error with no gRPC status keeps the subscription, and that a route stopped right after a failure does not subscribe again but does subscribe once when it is started again. - Unit tests of camel-dapr with the fix: 70 run, 0 failures, 0 skipped. - The formatter and import-sort plugins ran. I built offline with `-Dasm-version=9.11` because asm 9.11.1 was not in my local repository. # Target - [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch) # Tracking - [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it): [CAMEL-26044](https://issues.apache.org/jira/browse/CAMEL-26044). # Apache Camel coding standards and style - [x] I checked that each commit in the pull request has a meaningful subject line and body. - [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes. (I built camel-dapr and ran its tests, including the formatter and import-sort plugins, and copied the changed component page to its catalog copy. I did not run the full root build.) # AI-assisted contributions - [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used. This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer. _Claude Code on behalf of allthingssecurity_ 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
