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 d969c8d7df8f CAMEL-25614: camel-kubernetes - context reload on secret
or config map update stops after the watch is closed (#27801)
d969c8d7df8f is described below
commit d969c8d7df8f3b43e30cee95cabf969418dc8e2a
Author: allthingssecurity <[email protected]>
AuthorDate: Wed Oct 14 17:26:04 2026 +0530
CAMEL-25614: camel-kubernetes - context reload on secret or config map
update stops after the watch is closed (#27801)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../vault/ConfigmapsReloadTriggerTask.java | 82 ++++++++++-
.../secrets/vault/SecretsReloadTriggerTask.java | 82 ++++++++++-
...ConfigmapsReloadTriggerTaskWatchClosedTest.java | 158 +++++++++++++++++++++
.../SecretsReloadTriggerTaskWatchClosedTest.java | 158 +++++++++++++++++++++
4 files changed, 468 insertions(+), 12 deletions(-)
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 53c88cc96c9a..febf58b6f3c4 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
@@ -21,6 +21,7 @@ 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;
@@ -51,9 +52,15 @@ 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 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;
@@ -116,11 +123,64 @@ public class ConfigmapsReloadTriggerTask extends
ServiceSupport implements Camel
}
}
+ @Override
+ protected void doStop() throws Exception {
+ 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();
+ }
+ }
+
+ /**
+ * 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.
+ */
@Override
public void run() {
startingTime = Instant.now();
- final CountDownLatch isWatchClosed = new CountDownLatch(1);
- Watch watch =
kubernetesClient.configMaps().inNamespace(kubernetesClient.getNamespace()).watch(new
Watcher<>() {
+ 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);
+ }
+ }
+
+ 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()) {
@@ -144,18 +204,28 @@ public class ConfigmapsReloadTriggerTask extends
ServiceSupport implements Camel
@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();
}
});
+ }
- // Wait till watch gets closed
+ /**
+ * @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 {
- isWatchClosed.await();
+ if (isRunAllowed()) {
+ pause.await(delay, TimeUnit.MILLISECONDS);
+ }
} catch (InterruptedException e) {
- LOG.debug("Interrupted while waiting for the watch to close: {}",
e.getMessage());
Thread.currentThread().interrupt();
+ return false;
}
- watch.close();
+ 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 08db0f9fa210..bdac8ba41e67 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
@@ -21,6 +21,7 @@ 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.*;
@@ -48,9 +49,15 @@ 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 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;
@@ -113,11 +120,64 @@ public class SecretsReloadTriggerTask extends
ServiceSupport implements CamelCon
}
}
+ @Override
+ protected void doStop() throws Exception {
+ 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();
+ }
+ }
+
+ /**
+ * 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.
+ */
@Override
public void run() {
startingTime = Instant.now();
- final CountDownLatch isWatchClosed = new CountDownLatch(1);
- Watch watch =
kubernetesClient.secrets().inNamespace(kubernetesClient.getNamespace()).watch(new
Watcher<>() {
+ 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);
+ }
+ }
+
+ 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()) {
@@ -141,18 +201,28 @@ public class SecretsReloadTriggerTask extends
ServiceSupport implements CamelCon
@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();
}
});
+ }
- // Wait till watch gets closed
+ /**
+ * @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 {
- isWatchClosed.await();
+ if (isRunAllowed()) {
+ pause.await(delay, TimeUnit.MILLISECONDS);
+ }
} catch (InterruptedException e) {
- LOG.debug("Interrupted while waiting for the watch to close: {}",
e.getMessage());
Thread.currentThread().interrupt();
+ return false;
}
- watch.close();
+ return isRunAllowed();
}
protected boolean matchSecret(String name) {
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
new file mode 100644
index 000000000000..47db36080539
--- /dev/null
+++
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/config_maps/vault/ConfigmapsReloadTriggerTaskWatchClosedTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.config_maps.vault;
+
+import java.util.concurrent.CountDownLatch;
+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.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.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.
+ */
+@EnableKubernetesMockClient
+class ConfigmapsReloadTriggerTaskWatchClosedTest {
+
+ private static final String WATCH_PATH =
"/api/v1/namespaces/test/configmaps?allowWatchBookmarks=true&watch=true";
+
+ KubernetesMockServer server;
+ NamespacedKubernetesClient client;
+
+ @Test
+ void testWatchAgainAfterTheWatchIsClosedWithAnError() throws Exception {
+ expectWatchClosedWithGone();
+ expectWatchWithUpdate();
+
+ assertReloadIsTriggered();
+ assertEquals(2, server.getRequestCount());
+ }
+
+ @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());
+ }
+
+ /**
+ * A task that is stopped while it watches ends, so it does not keep the
thread that runs it, and it does not watch
+ * again.
+ */
+ @Test
+ void testStopEndsTheTaskThatWatches() throws Exception {
+ expectWatchWithUpdate();
+
+ CountDownLatch reloaded = new CountDownLatch(1);
+ AtomicReference<Thread> taskThread = new AtomicReference<>();
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new ConfigmapsReloadTriggerTask() {
+ @Override
+ public void run() {
+ taskThread.set(Thread.currentThread());
+ super.run();
+ }
+ }, reloaded);
+
+ // the task has a watch
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS), "the update of
the config map did not trigger a reload");
+ }
+
+ // the context is stopped, and the task with it
+ 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 schedule(CamelContext context, ConfigmapsReloadTriggerTask
task, CountDownLatch reloaded) throws Exception {
+ ConfigMapPropertiesFunction function = new
ConfigMapPropertiesFunction();
+ function.setCamelContext(context);
+ function.setClient(client);
+ context.getPropertiesComponent().addPropertiesFunction(function);
+
context.getVaultConfiguration().kubernetesConfigmaps().setConfigmaps("myconfigmap");
+ context.addService(new DefaultContextReloadStrategy() {
+ @Override
+ public void onReload(Object source) {
+ reloaded.countDown();
+ }
+ });
+ // scheduled the way camel-main does with
camel.vault.kubernetescm.refreshEnabled=true
+ PluginHelper.getPeriodTaskScheduler(context).scheduledTask(task);
+ context.start();
+ }
+}
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
new file mode 100644
index 000000000000..7de670f2a73e
--- /dev/null
+++
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/secrets/vault/SecretsReloadTriggerTaskWatchClosedTest.java
@@ -0,0 +1,158 @@
+/*
+ * 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.secrets.vault;
+
+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.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.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.
+ */
+@EnableKubernetesMockClient
+class SecretsReloadTriggerTaskWatchClosedTest {
+
+ private static final String WATCH_PATH =
"/api/v1/namespaces/test/secrets?allowWatchBookmarks=true&watch=true";
+
+ KubernetesMockServer server;
+ NamespacedKubernetesClient client;
+
+ @Test
+ void testWatchAgainAfterTheWatchIsClosedWithAnError() throws Exception {
+ expectWatchClosedWithGone();
+ expectWatchWithUpdate();
+
+ assertReloadIsTriggered();
+ assertEquals(2, server.getRequestCount());
+ }
+
+ @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());
+ }
+
+ /**
+ * A task that is stopped while it watches ends, so it does not keep the
thread that runs it, and it does not watch
+ * again.
+ */
+ @Test
+ void testStopEndsTheTaskThatWatches() throws Exception {
+ expectWatchWithUpdate();
+
+ CountDownLatch reloaded = new CountDownLatch(1);
+ AtomicReference<Thread> taskThread = new AtomicReference<>();
+ try (CamelContext context = new DefaultCamelContext()) {
+ schedule(context, new SecretsReloadTriggerTask() {
+ @Override
+ public void run() {
+ taskThread.set(Thread.currentThread());
+ super.run();
+ }
+ }, reloaded);
+
+ // the task has a watch
+ assertTrue(reloaded.await(20, TimeUnit.SECONDS), "the update of
the secret did not trigger a reload");
+ }
+
+ // the context is stopped, and the task with it
+ 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
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 void schedule(CamelContext context, SecretsReloadTriggerTask task,
CountDownLatch reloaded) throws Exception {
+ SecretPropertiesFunction function = new SecretPropertiesFunction();
+ function.setCamelContext(context);
+ function.setClient(client);
+ context.getPropertiesComponent().addPropertiesFunction(function);
+ context.getVaultConfiguration().kubernetes().setSecrets("mysecret");
+ context.addService(new DefaultContextReloadStrategy() {
+ @Override
+ public void onReload(Object source) {
+ reloaded.countDown();
+ }
+ });
+ // scheduled the way camel-main does with
camel.vault.kubernetes.refreshEnabled=true
+ PluginHelper.getPeriodTaskScheduler(context).scheduledTask(task);
+ context.start();
+ }
+}
