This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 3b0c32f0de7c CAMEL-25844: camel-dapr - subscribe again to
configuration changes when the stream ends (#28317)
3b0c32f0de7c is described below
commit 3b0c32f0de7cf27797d42b1a5af1ac2a60903f7b
Author: allthingssecurity <[email protected]>
AuthorDate: Fri Oct 23 13:52:06 2026 +0530
CAMEL-25844: camel-dapr - subscribe again to configuration changes when the
stream ends (#28317)
The Dapr configuration consumer subscribed once to the configuration stream
and stopped receiving changes when the stream failed or completed. It now
subscribes again, after one second and then with a growing delay of up to 30
seconds, while the consumer runs.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/dapr-component.adoc | 2 +
.../camel-dapr/src/main/docs/dapr-component.adoc | 2 +
.../dapr/consumer/DaprConfigurationConsumer.java | 24 +++++++++--
.../consumer/DaprConfigurationConsumerTest.java | 49 ++++++++++++++++++++++
4 files changed, 74 insertions(+), 3 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/dapr-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/dapr-component.adoc
index c43f3015a9cb..3f11fcc534fb 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/dapr-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/dapr-component.adoc
@@ -105,6 +105,8 @@ The Dapr Configuration Consumer in Apache Camel allows your
route to receive rea
To configure the consumer, the configStore name and configKeys must be
provided.
+When the subscription ends, for example because the Dapr sidecar restarts, the
consumer subscribes again, first after one second and then with a growing delay
of up to 30 seconds while the sidecar cannot be reached. A change made to the
configuration store while the consumer is not subscribed is not delivered.
+
== Examples
=== Producer Operations Examples
diff --git a/components/camel-dapr/src/main/docs/dapr-component.adoc
b/components/camel-dapr/src/main/docs/dapr-component.adoc
index c43f3015a9cb..3f11fcc534fb 100644
--- a/components/camel-dapr/src/main/docs/dapr-component.adoc
+++ b/components/camel-dapr/src/main/docs/dapr-component.adoc
@@ -105,6 +105,8 @@ The Dapr Configuration Consumer in Apache Camel allows your
route to receive rea
To configure the consumer, the configStore name and configKeys must be
provided.
+When the subscription ends, for example because the Dapr sidecar restarts, the
consumer subscribes again, first after one second and then with a growing delay
of up to 30 seconds while the sidecar cannot be reached. A change made to the
configuration store while the consumer is not subscribed is not delivered.
+
== Examples
=== Producer Operations Examples
diff --git
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumer.java
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumer.java
index 02da12ece5bb..23038a41ed3f 100644
---
a/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumer.java
+++
b/components/camel-dapr/src/main/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumer.java
@@ -36,11 +36,14 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
+import reactor.util.retry.Retry;
public class DaprConfigurationConsumer extends DefaultConsumer {
private static final Logger LOG =
LoggerFactory.getLogger(DaprConfigurationConsumer.class);
private static final Duration UNSUBSCRIBE_TIMEOUT = Duration.ofSeconds(10);
+ private static final Duration RESUBSCRIBE_DELAY = Duration.ofSeconds(1);
+ private static final Duration MAX_RESUBSCRIBE_DELAY =
Duration.ofSeconds(30);
private final String configStore;
private final List<String> configKeys;
private DaprClient client;
@@ -74,10 +77,25 @@ public class DaprConfigurationConsumer extends
DefaultConsumer {
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))
+ .retryWhen(Retry.backoff(Long.MAX_VALUE,
RESUBSCRIBE_DELAY).maxBackoff(MAX_RESUBSCRIBE_DELAY)
+ .transientErrors(true)
+ .doBeforeRetry(signal -> {
+ LOG.warn("The subscription to configuration
changes of store {} failed: {}. Subscribing again",
+ configStore,
signal.failure().getMessage());
+ LOG.debug("The subscription to configuration
changes of store {} failed", configStore,
+ signal.failure());
+ }));
subscription = responses.subscribe((response) -> {
- // every response contains the subscription id, the first ever
response has no items
+ // every response contains the subscription id, the first response
of a subscription has no items
subscriptionId = response.getSubscriptionId();
if (response.getItems() == null || response.getItems().isEmpty()) {
LOG.debug("App subscribed to config changes with subscription
id: {}", subscriptionId);
@@ -88,7 +106,7 @@ public class DaprConfigurationConsumer extends
DefaultConsumer {
AsyncCallback cb = defaultConsumerCallback(exchange, true);
getAsyncProcessor().process(exchange, cb);
}
- });
+ }, error -> getExceptionHandler().handleException("Error subscribing
to configuration changes", error));
}
@Override
diff --git
a/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumerTest.java
b/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumerTest.java
index 5070c41765d7..224db04fe53f 100644
---
a/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumerTest.java
+++
b/components/camel-dapr/src/test/java/org/apache/camel/component/dapr/consumer/DaprConfigurationConsumerTest.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.dapr.consumer;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import io.dapr.client.DaprClient;
import io.dapr.client.domain.ConfigurationItem;
@@ -49,6 +50,8 @@ import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -118,4 +121,50 @@ public class DaprConfigurationConsumerTest extends
CamelTestSupport {
// the client belongs to the endpoint: it is not closed
verify(mockClient, never()).close();
}
+
+ @Test
+ void testConfigurationConsumerSubscribesAgainWhenTheStreamFails() throws
Exception {
+ // the first stream ends with an error, as when the Dapr sidecar
restarts; the second one delivers an update
+ AtomicInteger calls = new AtomicInteger();
+ when(mockClient.subscribeConfiguration(any())).thenAnswer(inv ->
calls.incrementAndGet() == 1
+ ? Flux.just(new SubscribeConfigurationResponse("mySubId",
Map.of()))
+ .concatWith(Flux.error(new
IllegalStateException("UNAVAILABLE: sidecar restarted")))
+ : Flux.just(new SubscribeConfigurationResponse("mySubId2",
Map.of()), update("myVal2", "mySubId2"))
+ .concatWith(Flux.never()));
+ when(mockClient.unsubscribeConfiguration(anyString(), anyString()))
+ .thenReturn(Mono.just(new
UnsubscribeConfigurationResponse(true, "")));
+
+ consumer.doStart();
+
+ verify(mockClient,
timeout(10000).times(2)).subscribeConfiguration(any(SubscribeConfigurationRequest.class));
+ verify(processor, timeout(10000)).process(exchangeCaptor.capture(),
callbackCaptor.capture());
+ assertEquals(Map.of("myKey1", "myVal2"),
exchangeCaptor.getValue().getIn().getBody());
+
+ consumer.doStop();
+ verify(mockClient).unsubscribeConfiguration("mySubId2", "myStore");
+ }
+
+ @Test
+ void testConfigurationConsumerSubscribesAgainWhenTheStreamCompletes()
throws Exception {
+ // the first stream completes, as when the sidecar ends the call on
shutdown
+ AtomicInteger calls = new AtomicInteger();
+ when(mockClient.subscribeConfiguration(any())).thenAnswer(inv ->
calls.incrementAndGet() == 1
+ ? Flux.just(new SubscribeConfigurationResponse("mySubId",
Map.of()))
+ : Flux.just(update("myVal3",
"mySubId2")).concatWith(Flux.never()));
+ when(mockClient.unsubscribeConfiguration(anyString(), anyString()))
+ .thenReturn(Mono.just(new
UnsubscribeConfigurationResponse(true, "")));
+
+ consumer.doStart();
+
+ verify(processor, timeout(10000)).process(exchangeCaptor.capture(),
callbackCaptor.capture());
+ assertEquals(Map.of("myKey1", "myVal3"),
exchangeCaptor.getValue().getIn().getBody());
+
+ consumer.doStop();
+ verify(mockClient,
times(2)).subscribeConfiguration(any(SubscribeConfigurationRequest.class));
+ }
+
+ private static SubscribeConfigurationResponse update(String value, String
subscriptionId) {
+ return new SubscribeConfigurationResponse(
+ subscriptionId, Map.of("myKey1", new
ConfigurationItem("myKey1", value, "v")));
+ }
}
