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 59c7e033505b CAMEL-25653: camel-kubernetes - share the watch loop of
the secret and config map reload tasks, and use an informer (#29893)
59c7e033505b is described below
commit 59c7e033505b66f0055ea7bee252a08f0e753d16
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 15 01:06:05 2026 +0530
CAMEL-25653: camel-kubernetes - share the watch loop of the secret and
config map reload tasks, and use an informer (#29893)
…d config map reload tasks, and use an informer
SecretsReloadTriggerTask and ConfigmapsReloadTriggerTask had the same watch
loop, wait and stop logic since CAMEL-25614, and a third copy of the retry
delays of KubernetesHelper. They also missed an update made between the
close
of a watch and the next one: after 410 Gone the API server no longer has the
events since the resource version of the watch, and the new watch started
from
the current state.
The loop is now in KubernetesInformerLoop, which both tasks use. It runs a
fabric8 informer: it lists the resources, watches them, and lists them again
after 410 Gone, comparing the resource versions, so an update made while the
resources were not watched still triggers the reload. Only an update with a
new resource version of a resource whose name matches calls back; created
and deleted resources do not reload, as before, and the informer keeps only
the matching resources. When the informer cannot be started, or the client
stops it for good, a new one is started with the delays of KubernetesHelper
(1 second, doubling up to 30 seconds), whose constants are now public; it
starts from the resources of the previous one, so an update made in between
is not missed either. Stopping the task stops the informer and ends its
thread.
The informer lists the resources, so with RBAC the service account now
needs the list permission as well as watch: documented in the secrets and
config maps pages and in the 4.24 upgrade guide.
Tests: the reload tests of CAMEL-25614 now use the CRUD mock server with a
dispatcher that can hold a watch and close it with 410 Gone; a new test in
each updates the resource while the watch is held, and fails without this
change. KubernetesInformerLoopTest covers which events are updates, a new
informer after the client stopped the previous one, and stopping the loop.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../docs/kubernetes-config-maps-component.adoc | 5 +
.../catalog/docs/kubernetes-secrets-component.adoc | 5 +
.../docs/kubernetes-config-maps-component.adoc | 5 +
.../main/docs/kubernetes-secrets-component.adoc | 5 +
.../component/kubernetes/KubernetesHelper.java | 7 +-
.../kubernetes/KubernetesInformerLoop.java | 240 ++++++++++++++++++
.../vault/ConfigmapsReloadTriggerTask.java | 117 ++-------
.../secrets/vault/SecretsReloadTriggerTask.java | 116 ++-------
.../kubernetes/KubernetesInformerLoopTest.java | 239 ++++++++++++++++++
.../kubernetes/WatchControlDispatcher.java | 268 +++++++++++++++++++++
...ConfigmapsReloadTriggerTaskWatchClosedTest.java | 163 ++++++++-----
.../SecretsReloadTriggerTaskWatchClosedTest.java | 164 ++++++++-----
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 +
13 files changed, 1028 insertions(+), 319 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kubernetes-config-maps-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kubernetes-config-maps-component.adoc
index 669bfaedefb4..8d708e3e41dd 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kubernetes-config-maps-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kubernetes-config-maps-component.adoc
@@ -272,3 +272,8 @@ where `camel.vault.kubernetescm.refreshEnabled` will enable
the automatic contex
Whenever a configmap listed in the property, will be updated in the same
namespace of the running application, the Camel context will be reloaded,
refreshing the configmap value.
+Camel lists the ConfigMaps of that namespace and then watches them (with a
Kubernetes informer), so when RBAC is enabled the
+ServiceAccount needs the `list` and `watch` permissions on `configmaps`. When
the API server closes the watch (for example with
+410 Gone because the watch is too old), Camel lists the ConfigMaps again, so
an update made in the meantime still reloads the
+context.
+
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kubernetes-secrets-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kubernetes-secrets-component.adoc
index 56bcfde49274..618f91198fa7 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kubernetes-secrets-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/kubernetes-secrets-component.adoc
@@ -298,3 +298,8 @@ where `camel.vault.kubernetes.refreshEnabled` will enable
the automatic context
Whenever a secrets listed in the property, will be updated in the same
namespace of the running application, the Camel context will be reloaded,
refreshing the secret value.
+Camel lists the Secrets of that namespace and then watches them (with a
Kubernetes informer), so when RBAC is enabled the
+ServiceAccount needs the `list` and `watch` permissions on `secrets`. When the
API server closes the watch (for example with
+410 Gone because the watch is too old), Camel lists the Secrets again, so an
update made in the meantime still reloads the
+context.
+
diff --git
a/components/camel-kubernetes/src/main/docs/kubernetes-config-maps-component.adoc
b/components/camel-kubernetes/src/main/docs/kubernetes-config-maps-component.adoc
index 669bfaedefb4..8d708e3e41dd 100644
---
a/components/camel-kubernetes/src/main/docs/kubernetes-config-maps-component.adoc
+++
b/components/camel-kubernetes/src/main/docs/kubernetes-config-maps-component.adoc
@@ -272,3 +272,8 @@ where `camel.vault.kubernetescm.refreshEnabled` will enable
the automatic contex
Whenever a configmap listed in the property, will be updated in the same
namespace of the running application, the Camel context will be reloaded,
refreshing the configmap value.
+Camel lists the ConfigMaps of that namespace and then watches them (with a
Kubernetes informer), so when RBAC is enabled the
+ServiceAccount needs the `list` and `watch` permissions on `configmaps`. When
the API server closes the watch (for example with
+410 Gone because the watch is too old), Camel lists the ConfigMaps again, so
an update made in the meantime still reloads the
+context.
+
diff --git
a/components/camel-kubernetes/src/main/docs/kubernetes-secrets-component.adoc
b/components/camel-kubernetes/src/main/docs/kubernetes-secrets-component.adoc
index 56bcfde49274..618f91198fa7 100644
---
a/components/camel-kubernetes/src/main/docs/kubernetes-secrets-component.adoc
+++
b/components/camel-kubernetes/src/main/docs/kubernetes-secrets-component.adoc
@@ -298,3 +298,8 @@ where `camel.vault.kubernetes.refreshEnabled` will enable
the automatic context
Whenever a secrets listed in the property, will be updated in the same
namespace of the running application, the Camel context will be reloaded,
refreshing the secret value.
+Camel lists the Secrets of that namespace and then watches them (with a
Kubernetes informer), so when RBAC is enabled the
+ServiceAccount needs the `list` and `watch` permissions on `secrets`. When the
API server closes the watch (for example with
+410 Gone because the watch is too old), Camel lists the Secrets again, so an
update made in the meantime still reloads the
+context.
+
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/KubernetesHelper.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/KubernetesHelper.java
index 8afec60646b5..494025fb47de 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/KubernetesHelper.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/KubernetesHelper.java
@@ -94,15 +94,16 @@ public final class KubernetesHelper {
}
/**
- * The delay before a consumer watches again after the Kubernetes client
closed its watch with an error.
+ * The delay before a consumer, or a task that reloads the CamelContext,
watches again after the Kubernetes client
+ * closed its watch with an error.
*/
- static final long WATCH_AGAIN_DELAY_MILLIS = 1000;
+ public static final long WATCH_AGAIN_DELAY_MILLIS = 1000;
/**
* The maximum delay between two attempts to watch again, when watching
again keeps failing (for example while the
* API server cannot be reached).
*/
- static final long WATCH_AGAIN_MAX_DELAY_MILLIS = 30000;
+ public static final long WATCH_AGAIN_MAX_DELAY_MILLIS = 30000;
/**
* Creates the first watch of a consumer on its executor. If it fails (the
API server cannot be reached, answers 403
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/KubernetesInformerLoop.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/KubernetesInformerLoop.java
new file mode 100644
index 000000000000..5b378c8133f2
--- /dev/null
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/KubernetesInformerLoop.java
@@ -0,0 +1,240 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kubernetes;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import java.util.function.Predicate;
+
+import io.fabric8.kubernetes.api.model.HasMetadata;
+import io.fabric8.kubernetes.client.dsl.Informable;
+import io.fabric8.kubernetes.client.informers.ResourceEventHandler;
+import io.fabric8.kubernetes.client.informers.SharedIndexInformer;
+import io.fabric8.kubernetes.client.informers.cache.ReducedStateItemStore;
+import io.fabric8.kubernetes.client.utils.KubernetesSerialization;
+import org.apache.camel.support.service.ServiceSupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static
org.apache.camel.component.kubernetes.KubernetesHelper.WATCH_AGAIN_DELAY_MILLIS;
+import static
org.apache.camel.component.kubernetes.KubernetesHelper.WATCH_AGAIN_MAX_DELAY_MILLIS;
+
+/**
+ * Watches Kubernetes resources with an informer, until a service is stopped,
and calls back when a resource whose name
+ * matches is updated. Used by the tasks that reload the CamelContext when a
secret or a config map is updated.
+ * <p>
+ * The informer lists the resources, and then watches them. When the
Kubernetes client closes the watch, with 410 Gone
+ * because the resource version of the watch is too old (which happens to
long-running watches), or with any other
+ * error, the informer lists the resources again and compares them with the
ones it has, so an update made while the
+ * resources were not watched is not missed. Once started, the informer keeps
running: it retries a failed list or watch
+ * by itself, with its own backoff, and it is stopped only when this loop is
stopped. So an update it has queued for the
+ * handler, for example while a reload is running, is not dropped.
+ * <p>
+ * If the informer cannot be started (the API server cannot be reached,
answers 403 because the RBAC rules of the
+ * service account are not there yet, ...), a new one is started after
+ * {@link KubernetesHelper#WATCH_AGAIN_DELAY_MILLIS}, with a delay that
doubles up to
+ * {@link KubernetesHelper#WATCH_AGAIN_MAX_DELAY_MILLIS} while it cannot be
started.
+ * <p>
+ * The informer keeps only the name and the resource version of the resources
whose name matches, which is all it needs
+ * to tell an update. It does not keep the other resources of the namespace,
nor the data of the matching ones.
+ *
+ * @param <T> the type of the resources
+ */
+public final class KubernetesInformerLoop<T extends HasMetadata> {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(KubernetesInformerLoop.class);
+
+ private final ServiceSupport service;
+ private final String kind;
+ private final Class<T> type;
+ private final Informable<T> resources;
+ private final KubernetesSerialization serialization;
+ private final Predicate<String> nameFilter;
+ private final Consumer<T> onUpdate;
+ private volatile SharedIndexInformer<T> current;
+ private volatile boolean stopped;
+ // released when the informer is stopped, when the wait before the next
informer is over, and when the service is
+ // stopped
+ private volatile CountDownLatch released;
+
+ /**
+ * @param service the service that runs this loop, which ends when
the service is stopped
+ * @param kind the kind of the resources, for logging (Secrets,
ConfigMaps)
+ * @param type the type of the resources
+ * @param resources the resources to watch
+ * @param serialization the serialization of the Kubernetes client, to
keep the name and the resource version of the
+ * resources
+ * @param nameFilter whether a resource with this name is watched
+ * @param onUpdate called when a watched resource is updated
+ */
+ public KubernetesInformerLoop(ServiceSupport service, String kind,
Class<T> type, Informable<T> resources,
+ KubernetesSerialization serialization,
Predicate<String> nameFilter,
+ Consumer<T> onUpdate) {
+ this.service = service;
+ this.kind = kind;
+ this.type = type;
+ this.resources = resources;
+ this.serialization = serialization;
+ this.nameFilter = nameFilter;
+ this.onUpdate = onUpdate;
+ }
+
+ /**
+ * Watches the resources until the service, or this loop, is stopped. The
delay before starting a new informer is
+ * {@link KubernetesHelper#WATCH_AGAIN_DELAY_MILLIS} after the informer
could not be started the first time, and
+ * doubles while it cannot be started.
+ */
+ public void run() {
+ long delay = WATCH_AGAIN_DELAY_MILLIS;
+ while (isRunAllowed()) {
+ final CountDownLatch isInformerStopped = new CountDownLatch(1);
+ released = isInformerStopped;
+ SharedIndexInformer<T> informer = null;
+ try {
+ informer = resources.runnableInformer(0);
+ // Once started, the informer retries any error by itself, and
lists again after a watch error,
+ // instead of stopping for good after a watch error other than
410 Gone: a stopped informer drops the
+ // updates it has queued for the handler. An informer that
fails to start still stops, and this loop
+ // starts a new one.
+ informer.removeNamespaceIndex().itemStore(new
MatchingItemStore())
+ .exceptionHandler((started, t) -> started)
+ .addEventHandler(new UpdateHandler());
+ current = informer;
+ // a stop before the informer was published could not stop it
+ if (!isRunAllowed()) {
+ return;
+ }
+ // returns once the resources are listed and watched, and
fails if they cannot be
+ informer.run();
+ delay = WATCH_AGAIN_DELAY_MILLIS;
+ AtomicReference<Throwable> cause = new AtomicReference<>();
+ informer.stopped().whenComplete((r, e) -> {
+ cause.set(e);
+ isInformerStopped.countDown();
+ });
+ // Wait till the informer is stopped. A stop before the latch
was published could not release it
+ if (isRunAllowed()) {
+ isInformerStopped.await();
+ }
+ if (isRunAllowed()) {
+ // not expected, as a started informer stops only when
this loop is stopped
+ Throwable e = cause.get();
+ LOG.warn("The informer of Kubernetes {} stopped, watching
again in {} ms: {}", kind,
+ WATCH_AGAIN_DELAY_MILLIS, e != null ?
e.getMessage() : null);
+ }
+ } catch (InterruptedException e) {
+ LOG.debug("Interrupted while watching Kubernetes {}: {}",
kind, e.getMessage());
+ Thread.currentThread().interrupt();
+ return;
+ } catch (Exception e) {
+ if (!isRunAllowed()) {
+ return;
+ }
+ LOG.warn("Cannot watch Kubernetes {}, trying again in {} ms:
{}", kind, delay, e.getMessage());
+ LOG.debug("Cannot watch Kubernetes {}", kind, e);
+ } finally {
+ current = null;
+ if (informer != null) {
+ informer.stop();
+ }
+ }
+ if (!waitBeforeWatchingAgain(delay)) {
+ return;
+ }
+ delay = Math.min(delay * 2, WATCH_AGAIN_MAX_DELAY_MILLIS);
+ }
+ }
+
+ /**
+ * Stops the informer, and ends the wait before the next one, so the
thread that runs this loop ends. Called when
+ * the service is stopped.
+ */
+ public void stop() {
+ stopped = true;
+ SharedIndexInformer<T> informer = current;
+ if (informer != null) {
+ informer.stop();
+ }
+ CountDownLatch latch = released;
+ if (latch != null) {
+ latch.countDown();
+ }
+ }
+
+ private boolean isRunAllowed() {
+ return !stopped && service.isRunAllowed();
+ }
+
+ /**
+ * @return false if the service was stopped, or the thread interrupted,
while waiting
+ */
+ private boolean waitBeforeWatchingAgain(long delay) {
+ CountDownLatch pause = new CountDownLatch(1);
+ released = pause;
+ try {
+ if (isRunAllowed()) {
+ pause.await(delay, TimeUnit.MILLISECONDS);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ return isRunAllowed();
+ }
+
+ /**
+ * Keeps only the name and the resource version of the resources whose
name matches, so the informer keeps neither
+ * the other resources of the namespace, nor the data of the matching
ones. Each informer has its own store, so a
+ * new informer starts from what it lists, and an update it has not told
about cannot be hidden in the store of a
+ * previous one.
+ */
+ private final class MatchingItemStore extends ReducedStateItemStore<T> {
+
+ MatchingItemStore() {
+ super(ReducedStateItemStore.NAME_KEY_STATE, type, serialization);
+ }
+
+ @Override
+ public T put(String key, T obj) {
+ if (nameFilter.test(obj.getMetadata().getName())) {
+ return super.put(key, obj);
+ }
+ return null;
+ }
+ }
+
+ private final class UpdateHandler implements ResourceEventHandler<T> {
+
+ @Override
+ public void onAdd(T obj) {
+ // the resources that exist when the informer starts, and the new
ones, are not updates
+ }
+
+ @Override
+ public void onUpdate(T oldObj, T newObj) {
+ // the informer tells about a resource it lists again only when
its resource version changed
+ onUpdate.accept(newObj);
+ }
+
+ @Override
+ public void onDelete(T obj, boolean deletedFinalStateUnknown) {
+ // a deleted resource is not an update
+ }
+ }
+}
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTask.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTask.java
index febf58b6f3c4..b1eabfda5396 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTask.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTask.java
@@ -20,16 +20,12 @@ import java.time.Instant;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
import io.fabric8.kubernetes.api.model.ConfigMap;
import io.fabric8.kubernetes.client.KubernetesClient;
-import io.fabric8.kubernetes.client.Watch;
-import io.fabric8.kubernetes.client.Watcher;
-import io.fabric8.kubernetes.client.WatcherException;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
+import org.apache.camel.component.kubernetes.KubernetesInformerLoop;
import
org.apache.camel.component.kubernetes.properties.ConfigMapPropertiesFunction;
import org.apache.camel.spi.ContextReloadStrategy;
import org.apache.camel.spi.Metadata;
@@ -52,15 +48,10 @@ public class ConfigmapsReloadTriggerTask extends
ServiceSupport implements Camel
private KubernetesClient kubernetesClient;
private ConfigMapPropertiesFunction propertiesFunction;
private volatile Instant startingTime;
- // released when the current watch is closed with an error, when the wait
before the next watch is over, and when
- // this task is stopped
- private volatile CountDownLatch released;
+ private volatile KubernetesInformerLoop<ConfigMap> informerLoop;
private static final Logger LOG =
LoggerFactory.getLogger(ConfigmapsReloadTriggerTask.class);
- static final long WATCH_AGAIN_DELAY_MILLIS = 1000;
- static final long WATCH_AGAIN_MAX_DELAY_MILLIS = 30000;
-
@Override
public CamelContext getCamelContext() {
return camelContext;
@@ -107,6 +98,9 @@ public class ConfigmapsReloadTriggerTask extends
ServiceSupport implements Camel
}
kubernetesClient = propertiesFunction.getClient();
+ informerLoop = new KubernetesInformerLoop<>(
+ this, "ConfigMaps", ConfigMap.class,
kubernetesClient.configMaps().inNamespace(kubernetesClient.getNamespace()),
+ kubernetesClient.getKubernetesSerialization(),
this::matchSecret, this::onUpdate);
}
@Override
@@ -128,104 +122,33 @@ public class ConfigmapsReloadTriggerTask extends
ServiceSupport implements Camel
super.doStop();
// ends the watch, or the wait before the next watch, of the thread
that runs this task
- CountDownLatch latch = released;
- if (latch != null) {
- latch.countDown();
+ KubernetesInformerLoop<ConfigMap> loop = informerLoop;
+ if (loop != null) {
+ loop.stop();
}
}
/**
- * Watches the config maps until this task is stopped. This task is
submitted only once, and the Kubernetes client
- * closes a watch for good, with an exception, when the API server answers
410 Gone (the resource version of the
- * watch is too old, which can happen to a long-running watch), so the
config maps are watched again after
- * {@link #WATCH_AGAIN_DELAY_MILLIS}. A watch that cannot be created is
tried again with a delay that doubles up to
- * {@link #WATCH_AGAIN_MAX_DELAY_MILLIS}; the first of these delays is
{@link #WATCH_AGAIN_DELAY_MILLIS} when the
- * task starts and twice that after a closed watch.
+ * Watches the config maps until this task is stopped, with an informer
that lists them again when the Kubernetes
+ * client closes its watch (for example with 410 Gone, when the resource
version of the watch is too old), so an
+ * update made while the config maps were not watched is not missed. See
{@link KubernetesInformerLoop}.
*/
@Override
public void run() {
startingTime = Instant.now();
- long delay = WATCH_AGAIN_DELAY_MILLIS;
- while (isRunAllowed()) {
- final CountDownLatch isWatchClosed = new CountDownLatch(1);
- released = isWatchClosed;
- Watch watch = null;
- try {
- watch = watch(isWatchClosed);
- delay = WATCH_AGAIN_DELAY_MILLIS;
- // Wait till watch gets closed. A stop before the latch was
published could not release it
- if (isRunAllowed()) {
- isWatchClosed.await();
- }
- } catch (InterruptedException e) {
- LOG.debug("Interrupted while waiting for the watch to close:
{}", e.getMessage());
- Thread.currentThread().interrupt();
- return;
- } catch (Exception e) {
- if (!isRunAllowed()) {
- return;
- }
- LOG.warn("Cannot watch Kubernetes ConfigMaps, trying again in
{} ms: {}", delay, e.getMessage());
- LOG.debug("Cannot watch Kubernetes ConfigMaps", e);
- } finally {
- if (watch != null) {
- watch.close();
- }
- }
- if (!waitBeforeWatchingAgain(delay)) {
- return;
- }
- delay = Math.min(delay * 2, WATCH_AGAIN_MAX_DELAY_MILLIS);
- }
+ informerLoop.run();
}
- private Watch watch(CountDownLatch isWatchClosed) {
- return
kubernetesClient.configMaps().inNamespace(kubernetesClient.getNamespace()).watch(new
Watcher<>() {
- @Override
- public void eventReceived(Action action, ConfigMap configMap) {
- switch (action.name()) {
- case "MODIFIED":
- if (isReloadEnabled()) {
- if
(matchSecret(configMap.getMetadata().getName())) {
- LOG.info("Update for Kubernetes Configmaps: {}
detected, triggering CamelContext reload",
- configMap.getMetadata().getName());
- ContextReloadStrategy reload =
camelContext.hasService(ContextReloadStrategy.class);
- if (reload != null) {
- // trigger reload
- reload.onReload(this);
- }
- }
- }
- break;
- default:
- LOG.debug("Not watched event {}", action.name());
- }
- }
-
- @Override
- public void onClose(WatcherException e) {
- LOG.warn("The watch of Kubernetes ConfigMaps was closed,
watching again in {} ms: {}",
- WATCH_AGAIN_DELAY_MILLIS, e != null ? e.getMessage() :
null);
- isWatchClosed.countDown();
- }
- });
- }
-
- /**
- * @return false if this task was stopped, or the thread interrupted,
while waiting
- */
- private boolean waitBeforeWatchingAgain(long delay) {
- CountDownLatch pause = new CountDownLatch(1);
- released = pause;
- try {
- if (isRunAllowed()) {
- pause.await(delay, TimeUnit.MILLISECONDS);
+ private void onUpdate(ConfigMap configMap) {
+ if (isReloadEnabled()) {
+ LOG.info("Update for Kubernetes Configmaps: {} detected,
triggering CamelContext reload",
+ configMap.getMetadata().getName());
+ ContextReloadStrategy reload =
camelContext.hasService(ContextReloadStrategy.class);
+ if (reload != null) {
+ // trigger reload
+ reload.onReload(this);
}
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- return false;
}
- return isRunAllowed();
}
protected boolean matchSecret(String name) {
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTask.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTask.java
index bdac8ba41e67..76c073ae0a6f 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTask.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTask.java
@@ -20,13 +20,12 @@ import java.time.Instant;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
import io.fabric8.kubernetes.api.model.Secret;
-import io.fabric8.kubernetes.client.*;
+import io.fabric8.kubernetes.client.KubernetesClient;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
+import org.apache.camel.component.kubernetes.KubernetesInformerLoop;
import
org.apache.camel.component.kubernetes.properties.SecretPropertiesFunction;
import org.apache.camel.spi.ContextReloadStrategy;
import org.apache.camel.spi.Metadata;
@@ -49,15 +48,10 @@ public class SecretsReloadTriggerTask extends
ServiceSupport implements CamelCon
private KubernetesClient kubernetesClient;
private SecretPropertiesFunction propertiesFunction;
private volatile Instant startingTime;
- // released when the current watch is closed with an error, when the wait
before the next watch is over, and when
- // this task is stopped
- private volatile CountDownLatch released;
+ private volatile KubernetesInformerLoop<Secret> informerLoop;
private static final Logger LOG =
LoggerFactory.getLogger(SecretsReloadTriggerTask.class);
- static final long WATCH_AGAIN_DELAY_MILLIS = 1000;
- static final long WATCH_AGAIN_MAX_DELAY_MILLIS = 30000;
-
@Override
public CamelContext getCamelContext() {
return camelContext;
@@ -104,6 +98,9 @@ public class SecretsReloadTriggerTask extends ServiceSupport
implements CamelCon
}
kubernetesClient = propertiesFunction.getClient();
+ informerLoop = new KubernetesInformerLoop<>(
+ this, "Secrets", Secret.class,
kubernetesClient.secrets().inNamespace(kubernetesClient.getNamespace()),
+ kubernetesClient.getKubernetesSerialization(),
this::matchSecret, this::onUpdate);
}
@Override
@@ -125,104 +122,33 @@ public class SecretsReloadTriggerTask extends
ServiceSupport implements CamelCon
super.doStop();
// ends the watch, or the wait before the next watch, of the thread
that runs this task
- CountDownLatch latch = released;
- if (latch != null) {
- latch.countDown();
+ KubernetesInformerLoop<Secret> loop = informerLoop;
+ if (loop != null) {
+ loop.stop();
}
}
/**
- * Watches the secrets until this task is stopped. This task is submitted
only once, and the Kubernetes client
- * closes a watch for good, with an exception, when the API server answers
410 Gone (the resource version of the
- * watch is too old, which can happen to a long-running watch), so the
secrets are watched again after
- * {@link #WATCH_AGAIN_DELAY_MILLIS}. A watch that cannot be created is
tried again with a delay that doubles up to
- * {@link #WATCH_AGAIN_MAX_DELAY_MILLIS}; the first of these delays is
{@link #WATCH_AGAIN_DELAY_MILLIS} when the
- * task starts and twice that after a closed watch.
+ * Watches the secrets until this task is stopped, with an informer that
lists them again when the Kubernetes client
+ * closes its watch (for example with 410 Gone, when the resource version
of the watch is too old), so an update
+ * made while the secrets were not watched is not missed. See {@link
KubernetesInformerLoop}.
*/
@Override
public void run() {
startingTime = Instant.now();
- long delay = WATCH_AGAIN_DELAY_MILLIS;
- while (isRunAllowed()) {
- final CountDownLatch isWatchClosed = new CountDownLatch(1);
- released = isWatchClosed;
- Watch watch = null;
- try {
- watch = watch(isWatchClosed);
- delay = WATCH_AGAIN_DELAY_MILLIS;
- // Wait till watch gets closed. A stop before the latch was
published could not release it
- if (isRunAllowed()) {
- isWatchClosed.await();
- }
- } catch (InterruptedException e) {
- LOG.debug("Interrupted while waiting for the watch to close:
{}", e.getMessage());
- Thread.currentThread().interrupt();
- return;
- } catch (Exception e) {
- if (!isRunAllowed()) {
- return;
- }
- LOG.warn("Cannot watch Kubernetes Secrets, trying again in {}
ms: {}", delay, e.getMessage());
- LOG.debug("Cannot watch Kubernetes Secrets", e);
- } finally {
- if (watch != null) {
- watch.close();
- }
- }
- if (!waitBeforeWatchingAgain(delay)) {
- return;
- }
- delay = Math.min(delay * 2, WATCH_AGAIN_MAX_DELAY_MILLIS);
- }
+ informerLoop.run();
}
- private Watch watch(CountDownLatch isWatchClosed) {
- return
kubernetesClient.secrets().inNamespace(kubernetesClient.getNamespace()).watch(new
Watcher<>() {
- @Override
- public void eventReceived(Action action, Secret secret) {
- switch (action.name()) {
- case "MODIFIED":
- if (isReloadEnabled()) {
- if (matchSecret(secret.getMetadata().getName())) {
- LOG.info("Update for Kubernetes Secret: {}
detected, triggering CamelContext reload",
- secret.getMetadata().getName());
- ContextReloadStrategy reload =
camelContext.hasService(ContextReloadStrategy.class);
- if (reload != null) {
- // trigger reload
- reload.onReload(this);
- }
- }
- }
- break;
- default:
- LOG.debug("Not watched event {}", action.name());
- }
- }
-
- @Override
- public void onClose(WatcherException e) {
- LOG.warn("The watch of Kubernetes Secrets was closed, watching
again in {} ms: {}",
- WATCH_AGAIN_DELAY_MILLIS, e != null ? e.getMessage() :
null);
- isWatchClosed.countDown();
- }
- });
- }
-
- /**
- * @return false if this task was stopped, or the thread interrupted,
while waiting
- */
- private boolean waitBeforeWatchingAgain(long delay) {
- CountDownLatch pause = new CountDownLatch(1);
- released = pause;
- try {
- if (isRunAllowed()) {
- pause.await(delay, TimeUnit.MILLISECONDS);
+ private void onUpdate(Secret secret) {
+ if (isReloadEnabled()) {
+ LOG.info("Update for Kubernetes Secret: {} detected, triggering
CamelContext reload",
+ secret.getMetadata().getName());
+ ContextReloadStrategy reload =
camelContext.hasService(ContextReloadStrategy.class);
+ if (reload != null) {
+ // trigger reload
+ reload.onReload(this);
}
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- return false;
}
- return isRunAllowed();
}
protected boolean matchSecret(String name) {
diff --git
a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/KubernetesInformerLoopTest.java
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/KubernetesInformerLoopTest.java
new file mode 100644
index 000000000000..5b0ef35aafe8
--- /dev/null
+++
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/KubernetesInformerLoopTest.java
@@ -0,0 +1,239 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kubernetes;
+
+import java.time.Duration;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
+import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import org.apache.camel.support.service.ServiceSupport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The updates that {@link KubernetesInformerLoop} calls back for. The
informer calls the handler in the order of the
+ * events, so the first update it calls back for tells whether an earlier
event was taken as an update.
+ */
+class KubernetesInformerLoopTest {
+
+ private WatchControlDispatcher dispatcher;
+ private KubernetesMockServer server;
+ private NamespacedKubernetesClient client;
+ private final ServiceSupport service = new ServiceSupport() {
+ };
+ private final BlockingQueue<String> updates = new LinkedBlockingQueue<>();
+ private KubernetesInformerLoop<ConfigMap> loop;
+ private Thread thread;
+
+ @BeforeEach
+ void setUp() {
+ dispatcher = new
WatchControlDispatcher("/api/v1/namespaces/test/configmaps");
+ server = WatchControlDispatcher.createServer(dispatcher);
+ client = server.createClient();
+ create("my-a");
+ create("my-b");
+ create("other");
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ service.stop();
+ if (thread != null) {
+ loop.stop();
+ thread.join(20000);
+ }
+ client.close();
+ server.destroy();
+ }
+
+ @Test
+ void testOnlyAnUpdateOfAMatchingResourceIsAnUpdate() throws Exception {
+ start();
+ dispatcher.awaitOpenWatches(1);
+
+ update("other");
+ create("my-c");
+ client.configMaps().withName("my-a").delete();
+ update("my-b");
+
+ assertEquals("my-b", updates.poll(20, TimeUnit.SECONDS));
+ }
+
+ @Test
+ void testResourcesListedAgainAfterGoneAreUpdatesOnlyWhenChanged() throws
Exception {
+ dispatcher.holdFirstWatch();
+ start();
+ dispatcher.awaitOpenWatches(1);
+
+ // not sent to the held watch
+ update("my-b");
+ dispatcher.closeHeldWatchWithGone();
+ dispatcher.awaitOpenWatches(2);
+ update("my-a");
+
+ // my-a was listed again unchanged before it was updated
+ assertEquals("my-b", updates.poll(20, TimeUnit.SECONDS));
+ assertEquals("my-a", updates.poll(20, TimeUnit.SECONDS));
+ }
+
+ /**
+ * After a watch error other than 410 Gone, the informer lists and watches
the resources again by itself, and
+ * reports an update made in between.
+ */
+ @Test
+ void testUpdateMadeBeforeAWatchErrorOtherThanGoneIsReported() throws
Exception {
+ dispatcher.holdFirstWatch();
+ start();
+ dispatcher.awaitOpenWatches(1);
+
+ // not sent to the held watch
+ update("my-b");
+ dispatcher.sendInvalidObjectOnHeldWatch();
+ dispatcher.awaitOpenWatches(2);
+ update("my-a");
+
+ assertEquals("my-b", updates.poll(20, TimeUnit.SECONDS));
+ assertEquals("my-a", updates.poll(20, TimeUnit.SECONDS));
+ }
+
+ /**
+ * A watch error other than 410 Gone while the handler is busy, for
example with a reload, and an update is queued
+ * behind it: the informer is not stopped, so the queued update is not
dropped, and the busy handler is not
+ * interrupted. A stopped informer would drop the queued update, and a new
one would not report it, as the store
+ * already holds its resource version.
+ */
+ @Test
+ void testUpdateQueuedWhileTheHandlerIsBusyIsReportedAfterAWatchError()
throws Exception {
+ CountDownLatch inHandler = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ AtomicBoolean interrupted = new AtomicBoolean();
+ start(name -> {
+ if ("my-a".equals(name)) {
+ inHandler.countDown();
+ try {
+ release.await(20, TimeUnit.SECONDS);
+ } catch (InterruptedException e) {
+ interrupted.set(true);
+ Thread.currentThread().interrupt();
+ }
+ }
+ updates.add(name);
+ });
+ dispatcher.awaitOpenWatches(1);
+
+ update("my-a");
+ assertTrue(inHandler.await(20, TimeUnit.SECONDS), "my-a was not
reported");
+ // received by the informer, and queued behind the busy handler
+ update("my-b");
+ dispatcher.sendInvalidObjectOnLastOpenWatch();
+ // the Kubernetes client closes the watch while the handler is still
busy
+ dispatcher.awaitClosedWatches(1);
+ release.countDown();
+ // the informer lists and watches again (on the executor of its
handler, so once the handler is done)
+ dispatcher.awaitOpenWatches(2);
+
+ assertEquals("my-a", updates.poll(20, TimeUnit.SECONDS));
+ assertEquals("my-b", updates.poll(20, TimeUnit.SECONDS), "the queued
update of my-b was dropped");
+ assertFalse(interrupted.get(), "the busy handler was interrupted");
+ }
+
+ @Test
+ void testStopEndsTheLoop() throws Exception {
+ start();
+ dispatcher.awaitOpenWatches(1);
+ update("my-a");
+ assertEquals("my-a", updates.poll(20, TimeUnit.SECONDS));
+
+ service.stop();
+ loop.stop();
+
+ thread.join(20000);
+ assertFalse(thread.isAlive(), "the loop still runs after the service
was stopped");
+ }
+
+ /**
+ * The service starts a new loop when it is started again, so a loop that
is stopped ends even when the service runs
+ * again.
+ */
+ @Test
+ void testStopEndsTheLoopWhileTheServiceRuns() throws Exception {
+ start();
+ dispatcher.awaitOpenWatches(1);
+ update("my-a");
+ assertEquals("my-a", updates.poll(20, TimeUnit.SECONDS));
+
+ loop.stop();
+
+ thread.join(10000);
+ assertFalse(thread.isAlive(), "the stopped loop still runs");
+ }
+
+ /**
+ * A stop while the informer lists the resources, before it watches them,
ends the loop without watching.
+ */
+ @Test
+ void testStopWhileTheInformerStartsEndsTheLoop() throws Exception {
+ dispatcher.delayNextList(3);
+ start();
+ dispatcher.awaitListRequest();
+
+ service.stop();
+ loop.stop();
+
+ thread.join(20000);
+ assertFalse(thread.isAlive(), "the loop still runs after the service
was stopped");
+ // past the delay of the list, after which a watch would start
+ await().during(Duration.ofSeconds(4)).atMost(Duration.ofSeconds(10))
+ .untilAsserted(() -> assertEquals(0,
dispatcher.watchRequests(), "the stopped loop watched the resources"));
+ }
+
+ private void start() {
+ start(updates::add);
+ }
+
+ private void start(Consumer<String> onUpdate) {
+ service.start();
+ loop = new KubernetesInformerLoop<>(
+ service, "ConfigMaps", ConfigMap.class, client.configMaps(),
client.getKubernetesSerialization(),
+ name -> name.startsWith("my-"), c ->
onUpdate.accept(c.getMetadata().getName()));
+ thread = new Thread(loop::run, "informer-loop");
+ thread.start();
+ }
+
+ private void create(String name) {
+ client.configMaps().resource(new
ConfigMapBuilder().withNewMetadata().withName(name).endMetadata()
+ .addToData("greeting", "one").build()).create();
+ }
+
+ private void update(String name) {
+ client.configMaps().withName(name).edit(c -> new
ConfigMapBuilder(c).addToData("greeting", "two").build());
+ }
+}
diff --git
a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/WatchControlDispatcher.java
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/WatchControlDispatcher.java
new file mode 100644
index 000000000000..141e17b7b949
--- /dev/null
+++
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/WatchControlDispatcher.java
@@ -0,0 +1,268 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.kubernetes;
+
+import java.util.HashMap;
+import java.util.Queue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import io.fabric8.kubernetes.api.model.Status;
+import io.fabric8.kubernetes.api.model.StatusBuilder;
+import io.fabric8.kubernetes.api.model.WatchEvent;
+import io.fabric8.kubernetes.client.server.mock.KubernetesCrudDispatcher;
+import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import io.fabric8.kubernetes.client.utils.KubernetesSerialization;
+import io.fabric8.mockwebserver.Context;
+import io.fabric8.mockwebserver.MockWebServer;
+import io.fabric8.mockwebserver.http.ByteString;
+import io.fabric8.mockwebserver.http.MockResponse;
+import io.fabric8.mockwebserver.http.RecordedRequest;
+import io.fabric8.mockwebserver.http.Response;
+import io.fabric8.mockwebserver.http.WebSocket;
+import io.fabric8.mockwebserver.http.WebSocketListener;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A Kubernetes mock server dispatcher in CRUD mode that lets a test fail the
next requests for a kind of resources,
+ * hold the first watch of these resources (it then receives no event) and
close it with 410 Gone, and wait until a
+ * number of watches are open. A held watch that is closed with 410 Gone is
how the API server tells that the events
+ * since the resource version of the watch are gone: an update made in that
time is not sent to any watch.
+ */
+public class WatchControlDispatcher extends KubernetesCrudDispatcher {
+
+ private static final KubernetesSerialization SERIALIZATION = new
KubernetesSerialization();
+
+ private final String resourcesPath;
+ private final Queue<Integer> failures = new ConcurrentLinkedQueue<>();
+ private final AtomicInteger watches = new AtomicInteger();
+ private final Semaphore openWatches = new Semaphore(0);
+ private final Semaphore closedWatches = new Semaphore(0);
+ private final Semaphore listRequests = new Semaphore(0);
+ private final CompletableFuture<WebSocket> heldWatch = new
CompletableFuture<>();
+ private volatile boolean holdFirstWatch;
+ private volatile long nextListDelaySeconds;
+ private volatile WebSocket lastOpenWatch;
+
+ /**
+ * @param resourcesPath the path of the resources, for example
/api/v1/namespaces/test/secrets
+ */
+ public WatchControlDispatcher(String resourcesPath) {
+ this.resourcesPath = resourcesPath;
+ }
+
+ public static KubernetesMockServer createServer(WatchControlDispatcher
dispatcher) {
+ KubernetesMockServer server
+ = new KubernetesMockServer(new Context(), new MockWebServer(),
new HashMap<>(), dispatcher, false);
+ server.init();
+ return server;
+ }
+
+ /**
+ * The first watch of the resources is held: it receives no event until it
is closed with
+ * {@link #closeHeldWatchWithGone()}.
+ */
+ public void holdFirstWatch() {
+ holdFirstWatch = true;
+ }
+
+ /**
+ * The next request (list or watch) of the resources is answered with this
status code.
+ */
+ public void failNextRequest(int code) {
+ failures.add(code);
+ }
+
+ /**
+ * The answer to the next list request of the resources is delayed by the
given number of seconds.
+ */
+ public void delayNextList(long seconds) {
+ nextListDelaySeconds = seconds;
+ }
+
+ /**
+ * Waits until a list request of the resources is received.
+ */
+ public void awaitListRequest() throws InterruptedException {
+ assertTrue(listRequests.tryAcquire(20, TimeUnit.SECONDS), "the
resources were not listed");
+ }
+
+ /**
+ * The number of watch requests of the resources so far.
+ */
+ public int watchRequests() {
+ return watches.get();
+ }
+
+ /**
+ * Waits until the given number of watches of the resources have been
opened in total.
+ */
+ public void awaitOpenWatches(int count) throws InterruptedException {
+ // the permits are released once per open watch
+ assertTrue(openWatches.tryAcquire(count, 20, TimeUnit.SECONDS), "watch
" + count + " was not opened");
+ openWatches.release(count);
+ }
+
+ /**
+ * Waits until the given number of watches of the CRUD dispatcher (not the
held one) have been closed in total.
+ */
+ public void awaitClosedWatches(int count) throws InterruptedException {
+ assertTrue(closedWatches.tryAcquire(count, 20, TimeUnit.SECONDS),
"watch " + count + " was not closed");
+ closedWatches.release(count);
+ }
+
+ /**
+ * Closes the held first watch with 410 Gone, as the API server does when
the resource version of a watch is too
+ * old.
+ */
+ public void closeHeldWatchWithGone() throws Exception {
+ WebSocket webSocket = heldWatch.get(20, TimeUnit.SECONDS);
+ Status gone = new
StatusBuilder().withCode(410).withReason("Expired").withMessage("too old
resource version")
+ .build();
+ webSocket.send(SERIALIZATION.asJson(new WatchEvent(gone, "ERROR")));
+ }
+
+ /**
+ * Sends an event that is not a resource on the held first watch. The
Kubernetes client then closes the watch with
+ * an error other than 410 Gone. By default it would then stop the
informer for good.
+ */
+ public void sendInvalidObjectOnHeldWatch() throws Exception {
+ sendInvalidObject(heldWatch.get(20, TimeUnit.SECONDS));
+ }
+
+ /**
+ * Sends an event that is not a resource on the watch that was opened
last, after the events it was already sent.
+ * The Kubernetes client then closes the watch with an error other than
410 Gone.
+ */
+ public void sendInvalidObjectOnLastOpenWatch() {
+ WebSocket webSocket = lastOpenWatch;
+ assertNotNull(webSocket, "no watch is open");
+ sendInvalidObject(webSocket);
+ }
+
+ private static void sendInvalidObject(WebSocket webSocket) {
+ Status status = new StatusBuilder().withCode(500).withMessage("not a
resource").build();
+ webSocket.send(SERIALIZATION.asJson(new WatchEvent(status,
"MODIFIED")));
+ }
+
+ @Override
+ public MockResponse dispatch(RecordedRequest request) {
+ String path = request.getPath();
+ boolean resources = "GET".equals(request.getMethod())
+ && (path.equals(resourcesPath) ||
path.startsWith(resourcesPath + "?"));
+ if (!resources) {
+ return super.dispatch(request);
+ }
+ Integer code = failures.poll();
+ if (code != null) {
+ Status status = new
StatusBuilder().withCode(code).withMessage("failed by the test").build();
+ return new
MockResponse().setResponseCode(code).setBody(SERIALIZATION.asJson(status));
+ }
+ if (!path.contains("watch=true")) {
+ listRequests.release();
+ MockResponse response = super.dispatch(request);
+ long delay = nextListDelaySeconds;
+ if (delay > 0) {
+ nextListDelaySeconds = 0;
+ response.setBodyDelay(delay, TimeUnit.SECONDS);
+ }
+ return response;
+ }
+ if (watches.incrementAndGet() == 1 && holdFirstWatch) {
+ return new MockResponse().withWebSocketUpgrade(new
WebSocketListener() {
+ @Override
+ public void onOpen(WebSocket webSocket, Response response) {
+ heldWatch.complete(webSocket);
+ openWatches.release();
+ }
+ });
+ }
+ MockResponse response = super.dispatch(request);
+ WebSocketListener listener = response.getWebSocketListener();
+ if (listener != null) {
+ response.withWebSocketUpgrade(new OpenSignalingListener(listener));
+ }
+ return response;
+ }
+
+ /**
+ * Tells when a watch of the CRUD dispatcher is open, so an update made
after that is sent to it.
+ */
+ private final class OpenSignalingListener extends WebSocketListener {
+
+ private final WebSocketListener delegate;
+
+ OpenSignalingListener(WebSocketListener delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public void onBeforeAccept(WebSocket webSocket, Response response) {
+ delegate.onBeforeAccept(webSocket, response);
+ }
+
+ @Override
+ public void onOpen(WebSocket webSocket, Response response) {
+ delegate.onOpen(webSocket, response);
+ lastOpenWatch = webSocket;
+ openWatches.release();
+ }
+
+ @Override
+ public void onMessage(WebSocket webSocket, String text) {
+ delegate.onMessage(webSocket, text);
+ }
+
+ @Override
+ public void onMessage(WebSocket webSocket, ByteString bytes) {
+ delegate.onMessage(webSocket, bytes);
+ }
+
+ @Override
+ public void onMessage(WebSocket webSocket, byte[] bytes) {
+ delegate.onMessage(webSocket, bytes);
+ }
+
+ @Override
+ public void onPing(WebSocket webSocket, byte[] bytes) {
+ delegate.onPing(webSocket, bytes);
+ }
+
+ @Override
+ public void onClosing(WebSocket webSocket, int code, String reason) {
+ delegate.onClosing(webSocket, code, reason);
+ closedWatches.release();
+ }
+
+ @Override
+ public void onClosed(WebSocket webSocket, int code, String reason) {
+ delegate.onClosed(webSocket, code, reason);
+ }
+
+ @Override
+ public void onFailure(WebSocket webSocket, Throwable t, Response
response) {
+ delegate.onFailure(webSocket, t, response);
+ // a watch the client drops without a close frame
+ closedWatches.release();
+ }
+ }
+}
diff --git
a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTaskWatchClosedTest.java
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTaskWatchClosedTest.java
index 47db36080539..2c15e9157daf 100644
---
a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTaskWatchClosedTest.java
+++
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTaskWatchClosedTest.java
@@ -21,56 +21,125 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
-import io.fabric8.kubernetes.api.model.StatusBuilder;
-import io.fabric8.kubernetes.api.model.WatchEvent;
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
-import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.apache.camel.CamelContext;
+import org.apache.camel.component.kubernetes.WatchControlDispatcher;
import
org.apache.camel.component.kubernetes.properties.ConfigMapPropertiesFunction;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.support.DefaultContextReloadStrategy;
import org.apache.camel.support.PluginHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
- * The task that reloads the CamelContext when a config map is updated is
submitted once, and the Kubernetes client
- * closes its watch for good, with an exception, when the API server answers
410 Gone (the resource version of the watch
- * is too old, which can happen to a long-running watch). The task must then
watch again.
+ * The task that reloads the CamelContext when a config map is updated is
submitted once, and watches the config maps
+ * until it is stopped. The API server closes a watch with 410 Gone when the
resource version of the watch is too old
+ * (which can happen to a long-running watch), and then no longer has the
events since that version. The task must then
+ * watch again, and must not miss an update made while it was not watching.
*/
-@EnableKubernetesMockClient
class ConfigmapsReloadTriggerTaskWatchClosedTest {
- private static final String WATCH_PATH =
"/api/v1/namespaces/test/configmaps?allowWatchBookmarks=true&watch=true";
+ private WatchControlDispatcher dispatcher;
+ private KubernetesMockServer server;
+ private NamespacedKubernetesClient client;
+
+ @BeforeEach
+ void setUp() {
+ dispatcher = new
WatchControlDispatcher("/api/v1/namespaces/test/configmaps");
+ server = WatchControlDispatcher.createServer(dispatcher);
+ client = server.createClient();
+ client.configMaps().resource(new
ConfigMapBuilder().withNewMetadata().withName("myconfigmap").endMetadata()
+ .addToData("greeting", "one").build()).create();
+ }
+
+ @AfterEach
+ void tearDown() {
+ client.close();
+ server.destroy();
+ }
+
+ /**
+ * The API server closes a watch with 410 Gone when the events since its
resource version are gone, so an update
+ * made in that time is not sent to any watch.
+ */
+ @Test
+ void testUpdateMadeWhileTheWatchIsClosedIsNotMissed() throws Exception {
+ dispatcher.holdFirstWatch();
- KubernetesMockServer server;
- NamespacedKubernetesClient client;
+ CountDownLatch reloaded = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new ConfigmapsReloadTriggerTask(), reloaded);
+ dispatcher.awaitOpenWatches(1);
+
+ updateConfigMap("two");
+ dispatcher.closeHeldWatchWithGone();
+
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS),
+ "the update of the config map made while the watch was
closed did not trigger a reload");
+ }
+ }
@Test
void testWatchAgainAfterTheWatchIsClosedWithAnError() throws Exception {
- expectWatchClosedWithGone();
- expectWatchWithUpdate();
+ dispatcher.holdFirstWatch();
+
+ CountDownLatch reloaded = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new ConfigmapsReloadTriggerTask(), reloaded);
+ dispatcher.awaitOpenWatches(1);
+ dispatcher.closeHeldWatchWithGone();
+ dispatcher.awaitOpenWatches(2);
+
+ updateConfigMap("two");
+
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS),
+ "the update of the config map after the watch was closed
did not trigger a reload");
+ }
+ }
+
+ @Test
+ void testWatchAgainAfterAFailedAttemptToWatchAgain() throws Exception {
+ dispatcher.holdFirstWatch();
+
+ CountDownLatch reloaded = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new ConfigmapsReloadTriggerTask(), reloaded);
+ dispatcher.awaitOpenWatches(1);
+ // the first attempt to list or watch again fails
+ dispatcher.failNextRequest(403);
+ dispatcher.closeHeldWatchWithGone();
+ dispatcher.awaitOpenWatches(2);
+
+ updateConfigMap("two");
- assertReloadIsTriggered();
- assertEquals(2, server.getRequestCount());
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS),
+ "the update of the config map after a failed attempt to
watch again did not trigger a reload");
+ }
}
+ /**
+ * The config maps cannot be watched when the task starts, for example
while the RBAC rules of the service account
+ * are not there yet.
+ */
@Test
- void testWatchAgainAfterAFailedAttemptToWatch() throws Exception {
- expectWatchClosedWithGone();
- // the first attempt to watch again fails
- server.expect().withPath(WATCH_PATH)
- .andReturn(403, new
StatusBuilder().withCode(403).withReason("Forbidden").withMessage("forbidden").build())
- .once();
- expectWatchWithUpdate();
-
- // watched again after 1 second, failed, and watched again after 2
more seconds
- assertReloadIsTriggered();
- assertEquals(3, server.getRequestCount());
+ void testWatchAgainAfterAFailedFirstAttemptToWatch() throws Exception {
+ dispatcher.failNextRequest(403);
+
+ CountDownLatch reloaded = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new ConfigmapsReloadTriggerTask(), reloaded);
+ dispatcher.awaitOpenWatches(1);
+
+ updateConfigMap("two");
+
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS),
+ "the update of the config map after a failed first attempt
to watch did not trigger a reload");
+ }
}
/**
@@ -79,8 +148,6 @@ class ConfigmapsReloadTriggerTaskWatchClosedTest {
*/
@Test
void testStopEndsTheTaskThatWatches() throws Exception {
- expectWatchWithUpdate();
-
CountDownLatch reloaded = new CountDownLatch(1);
AtomicReference<Thread> taskThread = new AtomicReference<>();
try (CamelContext context = new DefaultCamelContext()) {
@@ -91,8 +158,10 @@ class ConfigmapsReloadTriggerTaskWatchClosedTest {
super.run();
}
}, reloaded);
+ dispatcher.awaitOpenWatches(1);
// the task has a watch
+ updateConfigMap("two");
assertTrue(reloaded.await(20, TimeUnit.SECONDS), "the update of
the config map did not trigger a reload");
}
@@ -100,43 +169,11 @@ class ConfigmapsReloadTriggerTaskWatchClosedTest {
Thread thread = taskThread.get();
thread.join(20000);
assertFalse(thread.isAlive(), "the stopped task still waits for its
watch to be closed");
- assertEquals(1, server.getRequestCount());
- }
-
- private void expectWatchClosedWithGone() {
- server.expect().withPath(WATCH_PATH)
- .andUpgradeToWebSocket()
- .open()
- .waitFor(10)
- .andEmit(new WatchEvent(
- new
StatusBuilder().withCode(410).withReason("Expired").withMessage("too old
resource version")
- .build(),
- "ERROR"))
- .done()
- .once();
- }
-
- private void expectWatchWithUpdate() {
- server.expect().withPath(WATCH_PATH)
- .andUpgradeToWebSocket()
- .open()
- .waitFor(10)
- .andEmit(new WatchEvent(
- new
ConfigMapBuilder().withNewMetadata().withName("myconfigmap").withNamespace("test")
-
.withResourceVersion("2").endMetadata().build(),
- "MODIFIED"))
- .done()
- .once();
}
- private void assertReloadIsTriggered() throws Exception {
- CountDownLatch reloaded = new CountDownLatch(1);
- try (CamelContext context = new DefaultCamelContext()) {
- schedule(context, new ConfigmapsReloadTriggerTask(), reloaded);
-
- assertTrue(reloaded.await(20, TimeUnit.SECONDS),
- "the update of the config map after the watch was closed
did not trigger a reload");
- }
+ private void updateConfigMap(String greeting) {
+ client.configMaps().withName("myconfigmap")
+ .edit(c -> new ConfigMapBuilder(c).addToData("greeting",
greeting).build());
}
private void schedule(CamelContext context, ConfigmapsReloadTriggerTask
task, CountDownLatch reloaded) throws Exception {
diff --git
a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTaskWatchClosedTest.java
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTaskWatchClosedTest.java
index 7de670f2a73e..fb43c2e14a3d 100644
---
a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTaskWatchClosedTest.java
+++
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTaskWatchClosedTest.java
@@ -16,61 +16,131 @@
*/
package org.apache.camel.component.kubernetes.secrets.vault;
+import java.util.Base64;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import io.fabric8.kubernetes.api.model.SecretBuilder;
-import io.fabric8.kubernetes.api.model.StatusBuilder;
-import io.fabric8.kubernetes.api.model.WatchEvent;
import io.fabric8.kubernetes.client.NamespacedKubernetesClient;
-import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
import org.apache.camel.CamelContext;
+import org.apache.camel.component.kubernetes.WatchControlDispatcher;
import
org.apache.camel.component.kubernetes.properties.SecretPropertiesFunction;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.support.DefaultContextReloadStrategy;
import org.apache.camel.support.PluginHelper;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
- * The task that reloads the CamelContext when a secret is updated is
submitted once, and the Kubernetes client closes
- * its watch for good, with an exception, when the API server answers 410 Gone
(the resource version of the watch is too
- * old, which can happen to a long-running watch). The task must then watch
again.
+ * The task that reloads the CamelContext when a secret is updated is
submitted once, and watches the secrets until it
+ * is stopped. The API server closes a watch with 410 Gone when the resource
version of the watch is too old (which can
+ * happen to a long-running watch), and then no longer has the events since
that version. The task must then watch
+ * again, and must not miss an update made while it was not watching.
*/
-@EnableKubernetesMockClient
class SecretsReloadTriggerTaskWatchClosedTest {
- private static final String WATCH_PATH =
"/api/v1/namespaces/test/secrets?allowWatchBookmarks=true&watch=true";
+ private WatchControlDispatcher dispatcher;
+ private KubernetesMockServer server;
+ private NamespacedKubernetesClient client;
+
+ @BeforeEach
+ void setUp() {
+ dispatcher = new
WatchControlDispatcher("/api/v1/namespaces/test/secrets");
+ server = WatchControlDispatcher.createServer(dispatcher);
+ client = server.createClient();
+ client.secrets().resource(new
SecretBuilder().withNewMetadata().withName("mysecret").endMetadata()
+ .addToData("password", encode("one")).build()).create();
+ }
- KubernetesMockServer server;
- NamespacedKubernetesClient client;
+ @AfterEach
+ void tearDown() {
+ client.close();
+ server.destroy();
+ }
+
+ /**
+ * The API server closes a watch with 410 Gone when the events since its
resource version are gone, so an update
+ * made in that time is not sent to any watch.
+ */
+ @Test
+ void testUpdateMadeWhileTheWatchIsClosedIsNotMissed() throws Exception {
+ dispatcher.holdFirstWatch();
+
+ CountDownLatch reloaded = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new SecretsReloadTriggerTask(), reloaded);
+ dispatcher.awaitOpenWatches(1);
+
+ updateSecret("two");
+ dispatcher.closeHeldWatchWithGone();
+
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS),
+ "the update of the secret made while the watch was closed
did not trigger a reload");
+ }
+ }
@Test
void testWatchAgainAfterTheWatchIsClosedWithAnError() throws Exception {
- expectWatchClosedWithGone();
- expectWatchWithUpdate();
+ dispatcher.holdFirstWatch();
+
+ CountDownLatch reloaded = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new SecretsReloadTriggerTask(), reloaded);
+ dispatcher.awaitOpenWatches(1);
+ dispatcher.closeHeldWatchWithGone();
+ dispatcher.awaitOpenWatches(2);
+
+ updateSecret("two");
+
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS),
+ "the update of the secret after the watch was closed did
not trigger a reload");
+ }
+ }
+
+ @Test
+ void testWatchAgainAfterAFailedAttemptToWatchAgain() throws Exception {
+ dispatcher.holdFirstWatch();
- assertReloadIsTriggered();
- assertEquals(2, server.getRequestCount());
+ CountDownLatch reloaded = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new SecretsReloadTriggerTask(), reloaded);
+ dispatcher.awaitOpenWatches(1);
+ // the first attempt to list or watch again fails
+ dispatcher.failNextRequest(403);
+ dispatcher.closeHeldWatchWithGone();
+ dispatcher.awaitOpenWatches(2);
+
+ updateSecret("two");
+
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS),
+ "the update of the secret after a failed attempt to watch
again did not trigger a reload");
+ }
}
+ /**
+ * The secrets cannot be watched when the task starts, for example while
the RBAC rules of the service account are
+ * not there yet.
+ */
@Test
- void testWatchAgainAfterAFailedAttemptToWatch() throws Exception {
- expectWatchClosedWithGone();
- // the first attempt to watch again fails
- server.expect().withPath(WATCH_PATH)
- .andReturn(403, new
StatusBuilder().withCode(403).withReason("Forbidden").withMessage("forbidden").build())
- .once();
- expectWatchWithUpdate();
-
- // watched again after 1 second, failed, and watched again after 2
more seconds
- assertReloadIsTriggered();
- assertEquals(3, server.getRequestCount());
+ void testWatchAgainAfterAFailedFirstAttemptToWatch() throws Exception {
+ dispatcher.failNextRequest(403);
+
+ CountDownLatch reloaded = new CountDownLatch(1);
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new SecretsReloadTriggerTask(), reloaded);
+ dispatcher.awaitOpenWatches(1);
+
+ updateSecret("two");
+
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS),
+ "the update of the secret after a failed first attempt to
watch did not trigger a reload");
+ }
}
/**
@@ -79,8 +149,6 @@ class SecretsReloadTriggerTaskWatchClosedTest {
*/
@Test
void testStopEndsTheTaskThatWatches() throws Exception {
- expectWatchWithUpdate();
-
CountDownLatch reloaded = new CountDownLatch(1);
AtomicReference<Thread> taskThread = new AtomicReference<>();
try (CamelContext context = new DefaultCamelContext()) {
@@ -91,8 +159,10 @@ class SecretsReloadTriggerTaskWatchClosedTest {
super.run();
}
}, reloaded);
+ dispatcher.awaitOpenWatches(1);
// the task has a watch
+ updateSecret("two");
assertTrue(reloaded.await(20, TimeUnit.SECONDS), "the update of
the secret did not trigger a reload");
}
@@ -100,43 +170,15 @@ class SecretsReloadTriggerTaskWatchClosedTest {
Thread thread = taskThread.get();
thread.join(20000);
assertFalse(thread.isAlive(), "the stopped task still waits for its
watch to be closed");
- assertEquals(1, server.getRequestCount());
}
- private void expectWatchClosedWithGone() {
- server.expect().withPath(WATCH_PATH)
- .andUpgradeToWebSocket()
- .open()
- .waitFor(10)
- .andEmit(new WatchEvent(
- new
StatusBuilder().withCode(410).withReason("Expired").withMessage("too old
resource version")
- .build(),
- "ERROR"))
- .done()
- .once();
+ private void updateSecret(String password) {
+ client.secrets().withName("mysecret")
+ .edit(s -> new SecretBuilder(s).addToData("password",
encode(password)).build());
}
- private void expectWatchWithUpdate() {
- server.expect().withPath(WATCH_PATH)
- .andUpgradeToWebSocket()
- .open()
- .waitFor(10)
- .andEmit(new WatchEvent(
- new
SecretBuilder().withNewMetadata().withName("mysecret").withNamespace("test")
-
.withResourceVersion("2").endMetadata().build(),
- "MODIFIED"))
- .done()
- .once();
- }
-
- private void assertReloadIsTriggered() throws Exception {
- CountDownLatch reloaded = new CountDownLatch(1);
- try (CamelContext context = new DefaultCamelContext()) {
- schedule(context, new SecretsReloadTriggerTask(), reloaded);
-
- assertTrue(reloaded.await(20, TimeUnit.SECONDS),
- "the update of the secret after the watch was closed did
not trigger a reload");
- }
+ private static String encode(String value) {
+ return Base64.getEncoder().encodeToString(value.getBytes());
}
private void schedule(CamelContext context, SecretsReloadTriggerTask task,
CountDownLatch reloaded) throws Exception {
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 386a3312bb40..33a70db846d5 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -373,6 +373,19 @@ assigns the scope to `admin-cli` and
`security-admin-console` only. When the `ke
pre-obtained `accessToken` must be issued with that scope too. Username and
password authentication uses `admin-cli`
and keeps working. The integration tests of camel-keycloak now run against
Keycloak 26.8.2.
+=== camel-kubernetes - the context reload on a secret or config map update
lists the resources
+
+With `camel.vault.kubernetes.refreshEnabled=true` or
`camel.vault.kubernetescm.refreshEnabled=true`, the task that
+reloads the CamelContext now uses a Kubernetes informer: it lists the Secrets
(or ConfigMaps) of the namespace and then
+watches them, and lists them again when the API server closes the watch (for
example with 410 Gone), so an update made
+while the watch was closed still reloads the context. Before, it only watched
them, and such an update was missed. When
+RBAC is enabled, the ServiceAccount now needs the `list` permission on
`secrets` (or `configmaps`) in addition to
+`watch`. The `list` permission does not widen what the ServiceAccount can
read: the watch the task used before had no
+resource version, so the API server already sent it every existing Secret (or
ConfigMap) of the namespace as an ADDED
+event. Without `list` the informer cannot start: the Kubernetes client logs an
ERROR, the task a WARN, and it tries again
+with a delay that doubles up to 30 seconds. Creating or deleting a secret or
config map still does not reload the
+context.
+
=== Java
*Breaking change:* Camel 4.24 requires *Java 21* as the minimum runtime
version.
