Repository: camel Updated Branches: refs/heads/camel-2.18.x b0e35cb47 -> 49adc55b7
CAMEL-11020: Camel Kubernetes consumers do not close watchers - Pods Consumer Project: http://git-wip-us.apache.org/repos/asf/camel/repo Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/c2add9fd Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/c2add9fd Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/c2add9fd Branch: refs/heads/camel-2.18.x Commit: c2add9fd951f6acbf49a30f0e1909e6f116861df Parents: b0e35cb Author: Andrea Cosentino <anco...@gmail.com> Authored: Thu Mar 16 11:46:49 2017 +0100 Committer: Andrea Cosentino <anco...@gmail.com> Committed: Thu Mar 16 13:08:58 2017 +0100 ---------------------------------------------------------------------- .../consumer/KubernetesPodsConsumer.java | 25 +++++++++++++++++--- 1 file changed, 22 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/camel/blob/c2add9fd/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java ---------------------------------------------------------------------- diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java index 24947b0..9abf29b 100644 --- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java +++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesPodsConsumer.java @@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneablePod; import io.fabric8.kubernetes.api.model.Pod; import io.fabric8.kubernetes.api.model.PodList; import io.fabric8.kubernetes.client.KubernetesClientException; +import io.fabric8.kubernetes.client.Watch; import io.fabric8.kubernetes.client.Watcher; import io.fabric8.kubernetes.client.dsl.ClientMixedOperation; import io.fabric8.kubernetes.client.dsl.ClientNonNamespaceOperation; @@ -43,6 +44,7 @@ public class KubernetesPodsConsumer extends DefaultConsumer { private final Processor processor; private ExecutorService executor; + private PodsConsumerTask podsWatcher; public KubernetesPodsConsumer(KubernetesEndpoint endpoint, Processor processor) { super(endpoint, processor); @@ -58,8 +60,9 @@ public class KubernetesPodsConsumer extends DefaultConsumer { protected void doStart() throws Exception { super.doStart(); executor = getEndpoint().createExecutor(); - - executor.submit(new PodsConsumerTask()); + + podsWatcher = new PodsConsumerTask(); + executor.submit(podsWatcher); } @Override @@ -69,8 +72,14 @@ public class KubernetesPodsConsumer extends DefaultConsumer { LOG.debug("Stopping Kubernetes Pods Consumer"); if (executor != null) { if (getEndpoint() != null && getEndpoint().getCamelContext() != null) { + if (podsWatcher != null) { + podsWatcher.getWatch().close(); + } getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor); } else { + if (podsWatcher != null) { + podsWatcher.getWatch().close(); + } executor.shutdownNow(); } } @@ -78,6 +87,8 @@ public class KubernetesPodsConsumer extends DefaultConsumer { } class PodsConsumerTask implements Runnable { + + private Watch watch; @Override public void run() { @@ -92,7 +103,7 @@ public class KubernetesPodsConsumer extends DefaultConsumer { if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getResourceName())) { w.withName(getEndpoint().getKubernetesConfiguration().getResourceName()); } - w.watch(new Watcher<Pod>() { + watch = w.watch(new Watcher<Pod>() { @Override public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action, @@ -117,6 +128,14 @@ public class KubernetesPodsConsumer extends DefaultConsumer { } }); + } + + public Watch getWatch() { + return watch; + } + + public void setWatch(Watch watch) { + this.watch = watch; } } }
