[1/5] camel git commit: CAMEL-11020: Camel Kubernetes consumers do not close watchers - Namespace Consumer

Previous Topic Next Topic
 
classic Classic list List threaded Threaded
5 messages Options
Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

[1/5] camel git commit: CAMEL-11020: Camel Kubernetes consumers do not close watchers - Namespace Consumer

acosentino
Repository: camel
Updated Branches:
  refs/heads/master 5846f22c3 -> f9e05f740


CAMEL-11020: Camel Kubernetes consumers do not close watchers - Namespace Consumer


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/99fca0ad
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/99fca0ad
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/99fca0ad

Branch: refs/heads/master
Commit: 99fca0add835adb2b532564d9d74e8d0349540ea
Parents: f617baa
Author: Andrea Cosentino <[hidden email]>
Authored: Thu Mar 16 11:50:56 2017 +0100
Committer: Andrea Cosentino <[hidden email]>
Committed: Thu Mar 16 13:20:46 2017 +0100

----------------------------------------------------------------------
 .../consumer/KubernetesNamespacesConsumer.java  | 23 ++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/99fca0ad/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java
index c2b4d45..901be16 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNamespacesConsumer.java
@@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneableNamespace;
 import io.fabric8.kubernetes.api.model.Namespace;
 import io.fabric8.kubernetes.api.model.NamespaceList;
 import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
 import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
@@ -42,6 +43,7 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
     
     private final Processor processor;
     private ExecutorService executor;
+    private NamespacesConsumerTask nsWatcher;
 
     public KubernetesNamespacesConsumer(KubernetesEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
@@ -58,7 +60,8 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
         super.doStart();
         executor = getEndpoint().createExecutor();
         
-        executor.submit(new NamespacesConsumerTask());
+        nsWatcher = new NamespacesConsumerTask();
+        executor.submit(nsWatcher);
     }
 
     @Override
@@ -68,8 +71,14 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
         LOG.debug("Stopping Kubernetes Namespace Consumer");
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null) {
+                if (nsWatcher != null) {
+                    nsWatcher.getWatch().close();
+                }
                 getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
             } else {
+                if (nsWatcher != null) {
+                    nsWatcher.getWatch().close();
+                }
                 executor.shutdownNow();
             }
         }
@@ -77,6 +86,8 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
     }
     
     class NamespacesConsumerTask implements Runnable {
+
+        private Watch watch;
         
         @Override
         public void run() {
@@ -84,7 +95,7 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
             if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getNamespace())) {
                 w.withName(getEndpoint().getKubernetesConfiguration().getNamespace());
             }
-            w.watch(new Watcher<Namespace>() {
+            watch = w.watch(new Watcher<Namespace>() {
 
                 @Override
                 public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action,
@@ -108,6 +119,14 @@ public class KubernetesNamespacesConsumer extends DefaultConsumer {
                     }                            
                 }
             });
+        }
+        
+        public Watch getWatch() {
+            return watch;
+        }
+
+        public void setWatch(Watch watch) {
+            this.watch = watch;
         }
     }
 }

Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

[2/5] camel git commit: CAMEL-11020: Camel Kubernetes consumers do not close watchers - Pods Consumer

acosentino
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/f617baaa
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f617baaa
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f617baaa

Branch: refs/heads/master
Commit: f617baaa1629e9a167e779324c8ed6630cbad791
Parents: 5846f22
Author: Andrea Cosentino <[hidden email]>
Authored: Thu Mar 16 11:46:49 2017 +0100
Committer: Andrea Cosentino <[hidden email]>
Committed: Thu Mar 16 13:20:46 2017 +0100

----------------------------------------------------------------------
 .../consumer/KubernetesPodsConsumer.java        | 25 +++++++++++++++++---
 1 file changed, 22 insertions(+), 3 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/f617baaa/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 6aa688e..76222c0 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.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.PodResource;
@@ -42,6 +43,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);
@@ -57,8 +59,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
@@ -68,8 +71,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();
             }
         }
@@ -77,6 +86,8 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
     }
 
     class PodsConsumerTask implements Runnable {
+
+        private Watch watch;
         
         @Override
         public void run() {
@@ -91,7 +102,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,
@@ -116,6 +127,14 @@ public class KubernetesPodsConsumer extends DefaultConsumer {
 
                 }
             });
+        }
+
+        public Watch getWatch() {
+            return watch;
+        }
+
+        public void setWatch(Watch watch) {
+            this.watch = watch;
         }
     }
 }

Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

[3/5] camel git commit: CAMEL-11020: Camel Kubernetes consumers do not close watchers - Nodes Consumer

acosentino
In reply to this post by acosentino
CAMEL-11020: Camel Kubernetes consumers do not close watchers - Nodes Consumer


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/d44034ad
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/d44034ad
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/d44034ad

Branch: refs/heads/master
Commit: d44034ad5a6c2874521408eb61223ccc818b381b
Parents: 99fca0a
Author: Andrea Cosentino <[hidden email]>
Authored: Thu Mar 16 11:56:42 2017 +0100
Committer: Andrea Cosentino <[hidden email]>
Committed: Thu Mar 16 13:20:47 2017 +0100

----------------------------------------------------------------------
 .../consumer/KubernetesNodesConsumer.java       | 23 ++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/d44034ad/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
index 14008a8..557c07b 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
@@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneableNode;
 import io.fabric8.kubernetes.api.model.Node;
 import io.fabric8.kubernetes.api.model.NodeList;
 import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
 import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
@@ -42,6 +43,7 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
 
     private final Processor processor;
     private ExecutorService executor;
+    private NodesConsumerTask nodesWatcher;
 
     public KubernetesNodesConsumer(KubernetesEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
@@ -58,7 +60,8 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
         super.doStart();
         executor = getEndpoint().createExecutor();
 
-        executor.submit(new NodesConsumerTask());
+        nodesWatcher = new NodesConsumerTask();
+        executor.submit(nodesWatcher);
     }
 
     @Override
@@ -68,8 +71,14 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
         LOG.debug("Stopping Kubernetes Nodes Consumer");
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null) {
+                if (nodesWatcher != null) {
+                 nodesWatcher.getWatch().close();
+                }
                 getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
             } else {
+                if (nodesWatcher != null) {
+                 nodesWatcher.getWatch().close();
+                }
                 executor.shutdownNow();
             }
         }
@@ -78,6 +87,8 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
 
     class NodesConsumerTask implements Runnable {
         
+        private Watch watch;
+        
         @Override
         public void run() {
             NonNamespaceOperation<Node, NodeList, DoneableNode, Resource<Node, DoneableNode>> w = getEndpoint().getKubernetesClient().nodes();
@@ -88,7 +99,7 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
             if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getResourceName())) {
                 w.withName(getEndpoint().getKubernetesConfiguration().getResourceName());
             }
-            w.watch(new Watcher<Node>() {
+            watch = w.watch(new Watcher<Node>() {
 
                 @Override
                 public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action,
@@ -113,6 +124,14 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
 
                 }
             });
+        }
+      
+        public Watch getWatch() {
+            return watch;
+        }
+
+        public void setWatch(Watch watch) {
+            this.watch = watch;
         }
     }
 }

Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

[4/5] camel git commit: CAMEL-11020: Camel Kubernetes consumers do not close watchers - RC consumer

acosentino
In reply to this post by acosentino
CAMEL-11020: Camel Kubernetes consumers do not close watchers - RC consumer


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/cad48310
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/cad48310
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/cad48310

Branch: refs/heads/master
Commit: cad4831054258bcfeaaaec5a6593ae33c0fa0e91
Parents: d44034a
Author: Andrea Cosentino <[hidden email]>
Authored: Thu Mar 16 12:14:43 2017 +0100
Committer: Andrea Cosentino <[hidden email]>
Committed: Thu Mar 16 13:20:47 2017 +0100

----------------------------------------------------------------------
 .../consumer/KubernetesNodesConsumer.java       |  4 +--
 ...ubernetesReplicationControllersConsumer.java | 26 +++++++++++++++++---
 2 files changed, 24 insertions(+), 6 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/cad48310/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
index 557c07b..4aa18c2 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesNodesConsumer.java
@@ -72,12 +72,12 @@ public class KubernetesNodesConsumer extends DefaultConsumer {
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null) {
                 if (nodesWatcher != null) {
-                 nodesWatcher.getWatch().close();
+                    nodesWatcher.getWatch().close();
                 }
                 getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
             } else {
                 if (nodesWatcher != null) {
-                 nodesWatcher.getWatch().close();
+                    nodesWatcher.getWatch().close();
                 }
                 executor.shutdownNow();
             }

http://git-wip-us.apache.org/repos/asf/camel/blob/cad48310/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
index e4f7700..9cb71da 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesReplicationControllersConsumer.java
@@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneableReplicationController;
 import io.fabric8.kubernetes.api.model.ReplicationController;
 import io.fabric8.kubernetes.api.model.ReplicationControllerList;
 import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.RollableScallableResource;
@@ -42,6 +43,7 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
 
     private final Processor processor;
     private ExecutorService executor;
+    private ReplicationControllersConsumerTask rcWatcher;
 
     public KubernetesReplicationControllersConsumer(KubernetesEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
@@ -58,8 +60,8 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
         super.doStart();
 
         executor = getEndpoint().createExecutor();
-
-        executor.submit(new ReplicationControllersConsumerTask());      
+        rcWatcher = new ReplicationControllersConsumerTask();
+        executor.submit(rcWatcher);      
     }
 
     @Override
@@ -69,8 +71,14 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
         LOG.debug("Stopping Kubernetes Replication Controllers Consumer");
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null) {
+                if (rcWatcher != null) {
+                    rcWatcher.getWatch().close();
+                }
                 getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
             } else {
+                if (rcWatcher != null) {
+                    rcWatcher.getWatch().close();
+                }
                 executor.shutdownNow();
             }
         }
@@ -78,7 +86,9 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
     }
     
     class ReplicationControllersConsumerTask implements Runnable {
-        
+        
+        private Watch watch;
+        
         @Override
         public void run() {
             MixedOperation<ReplicationController, ReplicationControllerList, DoneableReplicationController,
@@ -93,7 +103,7 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
             if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getResourceName())) {
                 w.withName(getEndpoint().getKubernetesConfiguration().getResourceName());
             }
-            w.watch(new Watcher<ReplicationController>() {
+            watch = w.watch(new Watcher<ReplicationController>() {
 
                 @Override
                 public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action,
@@ -120,5 +130,13 @@ public class KubernetesReplicationControllersConsumer extends DefaultConsumer {
 
             });
         }
+        
+        public Watch getWatch() {
+            return watch;
+        }
+
+        public void setWatch(Watch watch) {
+            this.watch = watch;
+        }
     }
 }

Reply | Threaded
Open this post in threaded view
|  
Report Content as Inappropriate

[5/5] camel git commit: CAMEL-11020: Camel Kubernetes consumers do not close watchers - Service consumer

acosentino
In reply to this post by acosentino
CAMEL-11020: Camel Kubernetes consumers do not close watchers - Service consumer


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/f9e05f74
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/f9e05f74
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/f9e05f74

Branch: refs/heads/master
Commit: f9e05f740587b00964e99aaab9c2f800f0c19dca
Parents: cad4831
Author: Andrea Cosentino <[hidden email]>
Authored: Thu Mar 16 13:06:58 2017 +0100
Committer: Andrea Cosentino <[hidden email]>
Committed: Thu Mar 16 13:20:47 2017 +0100

----------------------------------------------------------------------
 .../consumer/KubernetesServicesConsumer.java    | 23 ++++++++++++++++++--
 1 file changed, 21 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/f9e05f74/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java
index 1b87ab8..5353f76 100644
--- a/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java
+++ b/components/camel-kubernetes/src/main/java/org/apache/camel/component/kubernetes/consumer/KubernetesServicesConsumer.java
@@ -22,6 +22,7 @@ import io.fabric8.kubernetes.api.model.DoneableService;
 import io.fabric8.kubernetes.api.model.Service;
 import io.fabric8.kubernetes.api.model.ServiceList;
 import io.fabric8.kubernetes.client.KubernetesClientException;
+import io.fabric8.kubernetes.client.Watch;
 import io.fabric8.kubernetes.client.Watcher;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
@@ -42,6 +43,7 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
 
     private final Processor processor;
     private ExecutorService executor;
+    private ServicesConsumerTask servicesWatcher;
 
     public KubernetesServicesConsumer(KubernetesEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
@@ -58,7 +60,8 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
         super.doStart();
         executor = getEndpoint().createExecutor();
 
-        executor.submit(new ServicesConsumerTask());      
+        servicesWatcher = new ServicesConsumerTask();
+        executor.submit(servicesWatcher);      
 
     }
 
@@ -68,8 +71,14 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
         LOG.debug("Stopping Kubernetes Services Consumer");
         if (executor != null) {
             if (getEndpoint() != null && getEndpoint().getCamelContext() != null) {
+                if (servicesWatcher != null) {
+                    servicesWatcher.getWatch().close();
+                }
                 getEndpoint().getCamelContext().getExecutorServiceManager().shutdownNow(executor);
             } else {
+                if (servicesWatcher != null) {
+                    servicesWatcher.getWatch().close();
+                }
                 executor.shutdownNow();
             }
         }
@@ -78,6 +87,8 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
     
     class ServicesConsumerTask implements Runnable {
         
+        private Watch watch;
+        
         @Override
         public void run() {
             MixedOperation<Service, ServiceList, DoneableService, Resource<Service, DoneableService>> w = getEndpoint().getKubernetesClient().services();
@@ -91,7 +102,7 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
             if (ObjectHelper.isNotEmpty(getEndpoint().getKubernetesConfiguration().getResourceName())) {
                 w.withName(getEndpoint().getKubernetesConfiguration().getResourceName());
             }
-            w.watch(new Watcher<Service>() {
+            watch = w.watch(new Watcher<Service>() {
 
                 @Override
                 public void eventReceived(io.fabric8.kubernetes.client.Watcher.Action action,
@@ -118,5 +129,13 @@ public class KubernetesServicesConsumer extends DefaultConsumer {
 
             });
         }
+        
+        public Watch getWatch() {
+            return watch;
+        }
+
+        public void setWatch(Watch watch) {
+            this.watch = watch;
+        }
     }
 }

Loading...