gnodet-bot commented on code in PR #28317:
URL: https://github.com/apache/camel/pull/28317#discussion_r4355149098
##########
components/camel-dapr/src/main/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumer.java:
##########
@@ -88,7 +106,7 @@ protected void doStart() throws Exception {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
Review Comment:
๐ก **Observation: error callback is effectively unreachable**
`retryWhen(Retry.backoff(Long.MAX_VALUE, ...))` will retry `Long.MAX_VALUE`
(~9.2 ร 10ยนโธ) times before propagating the error downstream. In practice, this
error callback will never fire. That's not harmful โ it's good defensive
practice โ but worth noting so future maintainers don't rely on it for real
error handling or alerting. A comment explaining this would help.
##########
components/camel-dapr/src/main/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumer.java:
##########
@@ -74,10 +77,25 @@ protected void doStart() throws Exception {
client = getEndpoint().getClient();
SubscribeConfigurationRequest configRequest = new
SubscribeConfigurationRequest(configStore, configKeys);
- Flux<SubscribeConfigurationResponse> responses =
client.subscribeConfiguration(configRequest);
+ // the stream ends when the call to the sidecar fails or is closed
(for example when the sidecar restarts), and
+ // the Dapr client does not open it again: subscribe again (after a
delay) for as long as the consumer runs, else
+ // the consumer stays started and receives no further configuration
changes
+ Flux<SubscribeConfigurationResponse> responses = Flux.defer(() ->
client.subscribeConfiguration(configRequest))
+ .doOnComplete(() -> LOG.warn(
+ "The subscription to configuration changes of store {}
ended. Subscribing again in {} ms",
+ configStore, RESUBSCRIBE_DELAY.toMillis()))
+ .repeatWhen(completed ->
completed.delayElements(RESUBSCRIBE_DELAY))
Review Comment:
๐ก **Design note: asymmetric backoff between `repeatWhen` and `retryWhen`**
`repeatWhen` uses a fixed `RESUBSCRIBE_DELAY` (1 s), while `retryWhen` uses
exponential backoff up to `MAX_RESUBSCRIBE_DELAY` (30 s). This means normal
stream completions (e.g. clean sidecar shutdown) will always re-subscribe after
exactly 1 s, even if the sidecar is still coming up. If the sidecar completes
the stream rapidly in a restart loop, the consumer could end up in a tight
1-second polling cycle.
Consider applying the same exponential-backoff strategy to `repeatWhen`, for
example via `Flux.interval` with a growing delay, or at minimum document the
intentional asymmetry (the current docs do describe the 1 s โ 30 s behaviour,
but only for the error path).
--
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]
