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 688db726024e CAMEL-25857: camel-kubernetes - watch consumers try again
when their first watch cannot be created (#28380)
688db726024e is described below
commit 688db726024ed267d7a437521a8058bfbc52b5f1
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 24 19:13:41 2026 +0530
CAMEL-25857: camel-kubernetes - watch consumers try again when their first
watch cannot be created (#28380)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/kubernetes/KubernetesHelper.java | 32 +++++++-
.../config_maps/KubernetesConfigMapsConsumer.java | 2 +-
.../KubernetesCustomResourcesConsumer.java | 2 +-
.../deployments/KubernetesDeploymentsConsumer.java | 2 +-
.../events/KubernetesEventsConsumer.java | 2 +-
.../kubernetes/hpa/KubernetesHPAConsumer.java | 2 +-
.../namespaces/KubernetesNamespacesConsumer.java | 2 +-
.../kubernetes/nodes/KubernetesNodesConsumer.java | 2 +-
.../kubernetes/pods/KubernetesPodsConsumer.java | 2 +-
.../KubernetesReplicationControllersConsumer.java | 2 +-
.../services/KubernetesServicesConsumer.java | 2 +-
.../OpenshiftDeploymentConfigsConsumer.java | 2 +-
.../KubernetesPodsConsumerFirstWatchFailsTest.java | 96 ++++++++++++++++++++++
13 files changed, 138 insertions(+), 12 deletions(-)
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 c647781cfb2b..db7e3af04f26 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
@@ -104,6 +104,35 @@ public final class KubernetesHelper {
*/
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
+ * because the RBAC rules of the service account are not there yet, ...),
it is tried again as in
+ * {@link #watchAgain(ServiceSupport, ExecutorService, Runnable)}, until
it succeeds or the consumer is stopped.
+ *
+ * @param consumer the consumer of the watch
+ * @param executor the executor of the consumer
+ * @param task the task that creates the watch of the consumer
+ */
+ public static void watch(ServiceSupport consumer, ExecutorService
executor, Runnable task) {
+ executor.submit(() -> {
+ if (!consumer.isRunAllowed()) {
+ return;
+ }
+ try {
+ task.run();
+ } catch (Exception e) {
+ if (Thread.currentThread().isInterrupted() ||
!consumer.isRunAllowed()) {
+ LOG.debug("Failed to watch for {} as it is stopping",
consumer, e);
+ return;
+ }
+ // tried again until it succeeds: one line per attempt, the
stack trace at DEBUG
+ LOG.warn("Failed to watch for {}, trying again in {} ms: {}",
consumer, WATCH_AGAIN_DELAY_MILLIS, e.toString());
+ LOG.debug("Failed to watch for {}", consumer, e);
+ watchAgain(consumer, executor, task, WATCH_AGAIN_DELAY_MILLIS);
+ }
+ });
+ }
+
/**
* Watches again when the Kubernetes client closed the watch of a consumer
with an error. The client reconnects a
* watch by itself after transient errors, and only closes it with an
exception when it gives up: when the API
@@ -149,7 +178,8 @@ public final class KubernetesHelper {
}
long nextDelayMillis = Math.min(delayMillis * 2,
WATCH_AGAIN_MAX_DELAY_MILLIS);
LOG.warn("Failed to watch again for {}, trying again in {}
ms: {}", consumer, nextDelayMillis,
- e.getMessage(), e);
+ e.toString());
+ LOG.debug("Failed to watch again for {}", consumer, e);
watchAgain(consumer, executor, task, nextDelayMillis);
}
});
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/config_maps/KubernetesConfigMapsConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/config_maps/KubernetesConfigMapsConsumer.java
index 6fd6452f461c..155ead5d3604 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/config_maps/KubernetesConfigMapsConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/config_maps/KubernetesConfigMapsConsumer.java
@@ -60,7 +60,7 @@ public class KubernetesConfigMapsConsumer extends
DefaultConsumer {
executor = getEndpoint().createExecutor(this);
configMapWatcher = new ConfigMapsConsumerTask();
- executor.submit(configMapWatcher);
+ KubernetesHelper.watch(this, executor, configMapWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/customresources/KubernetesCustomResourcesConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/customresources/KubernetesCustomResourcesConsumer.java
index a429a2a12c4e..726b1a2fb113 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/customresources/KubernetesCustomResourcesConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/customresources/KubernetesCustomResourcesConsumer.java
@@ -59,7 +59,7 @@ public class KubernetesCustomResourcesConsumer extends
DefaultConsumer {
executor = getEndpoint().createExecutor(this);
customResourcesWatcher = new CustomResourcesConsumerTask();
- executor.submit(customResourcesWatcher);
+ KubernetesHelper.watch(this, executor, customResourcesWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/deployments/KubernetesDeploymentsConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/deployments/KubernetesDeploymentsConsumer.java
index 8a5ab50196df..1bde5a4ae972 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/deployments/KubernetesDeploymentsConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/deployments/KubernetesDeploymentsConsumer.java
@@ -60,7 +60,7 @@ public class KubernetesDeploymentsConsumer extends
DefaultConsumer {
executor = getEndpoint().createExecutor(this);
deploymentsWatcher = new DeploymentsConsumerTask();
- executor.submit(deploymentsWatcher);
+ KubernetesHelper.watch(this, executor, deploymentsWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/events/KubernetesEventsConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/events/KubernetesEventsConsumer.java
index 44e6474eed92..75016bac50dd 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/events/KubernetesEventsConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/events/KubernetesEventsConsumer.java
@@ -60,7 +60,7 @@ public class KubernetesEventsConsumer extends DefaultConsumer
{
executor = getEndpoint().createExecutor(this);
eventWatcher = new EventsConsumerTask();
- executor.submit(eventWatcher);
+ KubernetesHelper.watch(this, executor, eventWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/hpa/KubernetesHPAConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/hpa/KubernetesHPAConsumer.java
index e6b069f2e46b..9460f4ff83cc 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/hpa/KubernetesHPAConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/hpa/KubernetesHPAConsumer.java
@@ -60,7 +60,7 @@ public class KubernetesHPAConsumer extends DefaultConsumer {
executor = getEndpoint().createExecutor(this);
hpasWatcher = new HpaConsumerTask();
- executor.submit(hpasWatcher);
+ KubernetesHelper.watch(this, executor, hpasWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/namespaces/KubernetesNamespacesConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/namespaces/KubernetesNamespacesConsumer.java
index 300380459b72..e63153b182f6 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/namespaces/KubernetesNamespacesConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/namespaces/KubernetesNamespacesConsumer.java
@@ -59,7 +59,7 @@ public class KubernetesNamespacesConsumer extends
DefaultConsumer {
executor = getEndpoint().createExecutor(this);
nsWatcher = new NamespacesConsumerTask();
- executor.submit(nsWatcher);
+ KubernetesHelper.watch(this, executor, nsWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/nodes/KubernetesNodesConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/nodes/KubernetesNodesConsumer.java
index e75b8b366c78..9fa8a3751967 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/nodes/KubernetesNodesConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/nodes/KubernetesNodesConsumer.java
@@ -59,7 +59,7 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
executor = getEndpoint().createExecutor(this);
nodesWatcher = new NodesConsumerTask();
- executor.submit(nodesWatcher);
+ KubernetesHelper.watch(this, executor, nodesWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/pods/KubernetesPodsConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/pods/KubernetesPodsConsumer.java
index 2754f18dd0e9..6854b435ac0f 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/pods/KubernetesPodsConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/pods/KubernetesPodsConsumer.java
@@ -60,7 +60,7 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
executor = getEndpoint().createExecutor(this);
podsWatcher = new PodsConsumerTask();
- executor.submit(podsWatcher);
+ KubernetesHelper.watch(this, executor, podsWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/replication_controllers/KubernetesReplicationControllersConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/replication_controllers/KubernetesReplicationControllersConsumer.java
index 1549971544f0..fd40601606e7 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/replication_controllers/KubernetesReplicationControllersConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/replication_controllers/KubernetesReplicationControllersConsumer.java
@@ -60,7 +60,7 @@ public class KubernetesReplicationControllersConsumer extends
DefaultConsumer {
executor = getEndpoint().createExecutor(this);
rcWatcher = new ReplicationControllersConsumerTask();
- executor.submit(rcWatcher);
+ KubernetesHelper.watch(this, executor, rcWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/services/KubernetesServicesConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/services/KubernetesServicesConsumer.java
index 4a088ec31f53..9ebce834eb02 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/services/KubernetesServicesConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/services/KubernetesServicesConsumer.java
@@ -60,7 +60,7 @@ public class KubernetesServicesConsumer extends
DefaultConsumer {
executor = getEndpoint().createExecutor(this);
servicesWatcher = new ServicesConsumerTask();
- executor.submit(servicesWatcher);
+ KubernetesHelper.watch(this, executor, servicesWatcher);
}
diff --git
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/openshift/deploymentconfigs/OpenshiftDeploymentConfigsConsumer.java
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/openshift/deploymentconfigs/OpenshiftDeploymentConfigsConsumer.java
index 6ac0f8fc9a53..589f2665a7f1 100644
---
a/components/camel-kubernetes/src/main/java/org/apache/camel/component/openshift/deploymentconfigs/OpenshiftDeploymentConfigsConsumer.java
+++
b/components/camel-kubernetes/src/main/java/org/apache/camel/component/openshift/deploymentconfigs/OpenshiftDeploymentConfigsConsumer.java
@@ -61,7 +61,7 @@ public class OpenshiftDeploymentConfigsConsumer extends
DefaultConsumer {
executor = getEndpoint().createExecutor(this);
deploymentsWatcher = new DeploymentsConfigConsumerTask();
- executor.submit(deploymentsWatcher);
+ KubernetesHelper.watch(this, executor, deploymentsWatcher);
}
@Override
diff --git
a/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumerFirstWatchFailsTest.java
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumerFirstWatchFailsTest.java
new file mode 100644
index 000000000000..2997047f507e
--- /dev/null
+++
b/components/camel-kubernetes/src/test/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumerFirstWatchFailsTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.consumer;
+
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodBuilder;
+import io.fabric8.kubernetes.api.model.StatusBuilder;
+import io.fabric8.kubernetes.api.model.WatchEvent;
+import io.fabric8.kubernetes.client.KubernetesClient;
+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.BindToRegistry;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.kubernetes.KubernetesTestSupport;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * When the first watch of a consumer cannot be created (the API server cannot
be reached, or answers 403 because the
+ * RBAC rules of the service account are not there yet), the consumer must try
again, as it does when the client closes
+ * a watch with an error.
+ */
+@EnableKubernetesMockClient
+class KubernetesPodsConsumerFirstWatchFailsTest extends KubernetesTestSupport {
+
+ private static final String WATCH_PATH =
"/api/v1/namespaces/test/pods?allowWatchBookmarks=true&watch=true";
+
+ KubernetesMockServer server;
+ NamespacedKubernetesClient client;
+
+ @BindToRegistry("kubernetesClient")
+ public KubernetesClient getClient() {
+ return client;
+ }
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ void testWatchAgainWhenTheFirstWatchFails() throws Exception {
+ // the first watch fails
+ server.expect().withPath(WATCH_PATH)
+ .andReturn(403, new
StatusBuilder().withCode(403).withReason("Forbidden").withMessage("forbidden").build())
+ .once();
+ server.expect().withPath(WATCH_PATH)
+ .andUpgradeToWebSocket()
+ .open()
+ .waitFor(10)
+ .andEmit(new WatchEvent(
+ new
PodBuilder().withNewMetadata().withName("pod1").withNamespace("test")
+
.withResourceVersion("2").endMetadata().build(),
+ "ADDED"))
+ .done()
+ .once();
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMessageCount(1);
+ mock.expectedMessagesMatches(e -> "pod1".equals(
+ e.getMessage().getBody(Pod.class).getMetadata().getName()));
+
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("kubernetes-pods://kubernetes?kubernetesClient=#kubernetesClient&namespace=test").routeId("pods")
+ .to("mock:result");
+ }
+ });
+ context.start();
+
+ // watched again after 1 second
+ mock.setResultWaitTime(10000);
+ mock.assertIsSatisfied();
+ assertEquals(2, server.getRequestCount());
+
assertTrue(context.getRouteController().getRouteStatus("pods").isStarted());
+ }
+}
