From 44aab8cfd406d5c0752fa8cf79533d218c634a61 Mon Sep 17 00:00:00 2001
From: erabii <eugen.rabii@gmail.com>
Date: Tue, 22 Nov 2022 03:32:02 +0200
Subject: [PATCH] Fix 1139 (#1140)

---
 .../kubernetes/fabric8/Fabric8Utils.java      |  88 ++++++
 .../kubernetes/fabric8/Fabric8UtilsTests.java |  81 +++++
 ...Fabric8ConfigMapPropertySourceLocator.java |   2 +-
 .../fabric8/config/Fabric8ConfigUtils.java    |  52 +---
 .../Fabric8SecretsPropertySourceLocator.java  |   2 +-
 .../config/Fabric8ConfigUtilsTests.java       |  48 ---
 .../discovery/KubernetesCatalogWatch.java     |  25 +-
 ...bernetesCatalogWatchAutoConfiguration.java |   6 +-
 .../Fabric8KubernetesCatalogWatchTests.java   | 282 ++++++++++++++++++
 .../discovery/KubernetesCatalogWatchTest.java | 135 ++++++---
 10 files changed, 568 insertions(+), 153 deletions(-)
 create mode 100644 spring-cloud-kubernetes-fabric8-autoconfig/src/main/java/org/springframework/cloud/kubernetes/fabric8/Fabric8Utils.java
 create mode 100644 spring-cloud-kubernetes-fabric8-autoconfig/src/test/java/org/springframework/cloud/kubernetes/fabric8/Fabric8UtilsTests.java
 create mode 100644 spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/Fabric8KubernetesCatalogWatchTests.java

diff --git a/spring-cloud-kubernetes-fabric8-autoconfig/src/main/java/org/springframework/cloud/kubernetes/fabric8/Fabric8Utils.java b/spring-cloud-kubernetes-fabric8-autoconfig/src/main/java/org/springframework/cloud/kubernetes/fabric8/Fabric8Utils.java
new file mode 100644
index 00000000..2aa5fa20
--- /dev/null
+++ b/spring-cloud-kubernetes-fabric8-autoconfig/src/main/java/org/springframework/cloud/kubernetes/fabric8/Fabric8Utils.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.kubernetes.fabric8;
+
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
+import org.springframework.cloud.kubernetes.commons.config.NamespaceResolutionFailedException;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.util.StringUtils;
+
+/**
+ * Utility class related to Fabric8 Client. It resides in this module because it is
+ * supposed to be re-used in fabric8 specific ones (like config or discovery)
+ *
+ * @author wind57
+ */
+public final class Fabric8Utils {
+
+	private Fabric8Utils() {
+
+	}
+
+	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(Fabric8Utils.class));
+
+	/**
+	 * this method does the namespace resolution for both config map and secrets
+	 * implementations. It tries these places to find the namespace:
+	 *
+	 * <pre>
+	 *     1. from a normalized source (which can be null)
+	 *     2. from a property 'spring.cloud.kubernetes.client.namespace', if such is present
+	 *     3. from a String residing in a file denoted by `spring.cloud.kubernetes.client.serviceAccountNamespacePath`
+	 * 	      property, if such is present
+	 * 	   4. from a String residing in `/var/run/secrets/kubernetes.io/serviceaccount/namespace` file,
+	 * 	  	  if such is present (kubernetes default path)
+	 * 	   5. from KubernetesClient::getNamespace, which is implementation specific.
+	 * </pre>
+	 *
+	 * If any of the above fail, we throw a {@link NamespaceResolutionFailedException}.
+	 * @param namespace normalized namespace
+	 * @param configurationTarget Config Map/Secret
+	 * @param provider the provider which computes the namespace
+	 * @param client fabric8 Kubernetes client
+	 * @return application namespace
+	 * @throws NamespaceResolutionFailedException when namespace could not be resolved
+	 */
+	public static String getApplicationNamespace(KubernetesClient client, String namespace, String configurationTarget,
+			KubernetesNamespaceProvider provider) {
+
+		if (StringUtils.hasText(namespace)) {
+			LOG.debug(configurationTarget + " namespace : " + namespace);
+			return namespace;
+		}
+
+		if (provider != null) {
+			String providerNamespace = provider.getNamespace();
+			if (StringUtils.hasText(providerNamespace)) {
+				LOG.debug(() -> configurationTarget + " namespace from provider : " + providerNamespace);
+				return providerNamespace;
+			}
+		}
+
+		String clientNamespace = client.getNamespace();
+		LOG.debug(() -> configurationTarget + " namespace from client : " + clientNamespace);
+		if (clientNamespace == null) {
+			throw new NamespaceResolutionFailedException("unresolved namespace");
+		}
+		return clientNamespace;
+
+	}
+
+}
diff --git a/spring-cloud-kubernetes-fabric8-autoconfig/src/test/java/org/springframework/cloud/kubernetes/fabric8/Fabric8UtilsTests.java b/spring-cloud-kubernetes-fabric8-autoconfig/src/test/java/org/springframework/cloud/kubernetes/fabric8/Fabric8UtilsTests.java
new file mode 100644
index 00000000..1cd3cef3
--- /dev/null
+++ b/spring-cloud-kubernetes-fabric8-autoconfig/src/test/java/org/springframework/cloud/kubernetes/fabric8/Fabric8UtilsTests.java
@@ -0,0 +1,81 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.kubernetes.fabric8;
+
+import io.fabric8.kubernetes.client.DefaultKubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
+import org.springframework.cloud.kubernetes.commons.config.NamespaceResolutionFailedException;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * @author wind57
+ */
+@EnableKubernetesMockClient(crud = true, https = false)
+class Fabric8UtilsTests {
+
+	private KubernetesClient client;
+
+	private final DefaultKubernetesClient mockClient = Mockito.mock(DefaultKubernetesClient.class);
+
+	private final KubernetesNamespaceProvider provider = Mockito.mock(KubernetesNamespaceProvider.class);
+
+	@Test
+	void testGetApplicationNamespaceNotPresent() {
+		String result = Fabric8Utils.getApplicationNamespace(client, "", "target", null);
+		assertThat(result).isEqualTo("test");
+	}
+
+	@Test
+	void testGetApplicationNamespacePresent() {
+		String result = Fabric8Utils.getApplicationNamespace(client, "namespace", "target", null);
+		assertThat(result).isEqualTo("namespace");
+	}
+
+	@Test
+	void testNamespaceFromNormalizedSource() {
+		String result = Fabric8Utils.getApplicationNamespace(client, "abc", "target", null);
+		assertThat(result).isEqualTo("abc");
+	}
+
+	@Test
+	void testNamespaceFromProvider() {
+		Mockito.when(provider.getNamespace()).thenReturn("def");
+		String result = Fabric8Utils.getApplicationNamespace(client, "", "target", provider);
+		assertThat(result).isEqualTo("def");
+	}
+
+	@Test
+	void testNamespaceFromClient() {
+		Mockito.when(mockClient.getNamespace()).thenReturn("qwe");
+		String result = Fabric8Utils.getApplicationNamespace(mockClient, "", "target", null);
+		assertThat(result).isEqualTo("qwe");
+	}
+
+	@Test
+	void testNamespaceResolutionFailed() {
+		assertThatThrownBy(() -> Fabric8Utils.getApplicationNamespace(mockClient, "", "target", null))
+				.isInstanceOf(NamespaceResolutionFailedException.class);
+	}
+
+}
diff --git a/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigMapPropertySourceLocator.java b/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigMapPropertySourceLocator.java
index f76a4c61..b8e9a70d 100644
--- a/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigMapPropertySourceLocator.java
+++ b/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigMapPropertySourceLocator.java
@@ -27,7 +27,7 @@ import org.springframework.core.annotation.Order;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.core.env.MapPropertySource;
 
-import static org.springframework.cloud.kubernetes.fabric8.config.Fabric8ConfigUtils.getApplicationNamespace;
+import static org.springframework.cloud.kubernetes.fabric8.Fabric8Utils.getApplicationNamespace;
 
 /**
  * A {@link PropertySourceLocator} that uses config maps.
diff --git a/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtils.java b/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtils.java
index e09f6c7d..7fcab208 100644
--- a/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtils.java
+++ b/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtils.java
@@ -31,11 +31,10 @@ import org.apache.commons.logging.LogFactory;
 import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.MultipleSourcesContainer;
-import org.springframework.cloud.kubernetes.commons.config.NamespaceResolutionFailedException;
 import org.springframework.cloud.kubernetes.commons.config.StrippedSourceContainer;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties;
+import org.springframework.cloud.kubernetes.fabric8.Fabric8Utils;
 import org.springframework.core.env.Environment;
-import org.springframework.util.StringUtils;
 
 /**
  * Utility class that works with configuration properties.
@@ -56,59 +55,12 @@ public final class Fabric8ConfigUtils {
 			ConfigReloadProperties properties, String target) {
 		Set<String> namespaces = properties.namespaces();
 		if (namespaces.isEmpty()) {
-			namespaces = Set.of(getApplicationNamespace(client, null, target, provider));
+			namespaces = Set.of(Fabric8Utils.getApplicationNamespace(client, null, target, provider));
 		}
 		LOG.debug("informer namespaces : " + namespaces);
 		return namespaces;
 	}
 
-	/**
-	 * this method does the namespace resolution for both config map and secrets
-	 * implementations. It tries these places to find the namespace:
-	 *
-	 * <pre>
-	 *     1. from a normalized source (which can be null)
-	 *     2. from a property 'spring.cloud.kubernetes.client.namespace', if such is present
-	 *     3. from a String residing in a file denoted by `spring.cloud.kubernetes.client.serviceAccountNamespacePath`
-	 * 	      property, if such is present
-	 * 	   4. from a String residing in `/var/run/secrets/kubernetes.io/serviceaccount/namespace` file,
-	 * 	  	  if such is present (kubernetes default path)
-	 * 	   5. from KubernetesClient::getNamespace, which is implementation specific.
-	 * </pre>
-	 *
-	 * If any of the above fail, we throw a NamespaceResolutionFailedException.
-	 * @param namespace normalized namespace
-	 * @param configurationTarget Config Map/Secret
-	 * @param provider the provider which computes the namespace
-	 * @param client fabric8 Kubernetes client
-	 * @return application namespace
-	 * @throws NamespaceResolutionFailedException when namespace could not be resolved
-	 */
-	static String getApplicationNamespace(KubernetesClient client, String namespace, String configurationTarget,
-			KubernetesNamespaceProvider provider) {
-
-		if (StringUtils.hasText(namespace)) {
-			LOG.debug(configurationTarget + " namespace : " + namespace);
-			return namespace;
-		}
-
-		if (provider != null) {
-			String providerNamespace = provider.getNamespace();
-			if (StringUtils.hasText(providerNamespace)) {
-				LOG.debug(configurationTarget + " namespace from provider : " + providerNamespace);
-				return providerNamespace;
-			}
-		}
-
-		String clientNamespace = client.getNamespace();
-		LOG.debug(configurationTarget + " namespace from client : " + clientNamespace);
-		if (clientNamespace == null) {
-			throw new NamespaceResolutionFailedException("unresolved namespace");
-		}
-		return clientNamespace;
-
-	}
-
 	/**
 	 * <pre>
 	 *     1. read all secrets in the provided namespace
diff --git a/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8SecretsPropertySourceLocator.java b/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8SecretsPropertySourceLocator.java
index 98d816e5..829648fb 100644
--- a/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8SecretsPropertySourceLocator.java
+++ b/spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8SecretsPropertySourceLocator.java
@@ -27,7 +27,7 @@ import org.springframework.core.annotation.Order;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.core.env.MapPropertySource;
 
-import static org.springframework.cloud.kubernetes.fabric8.config.Fabric8ConfigUtils.getApplicationNamespace;
+import static org.springframework.cloud.kubernetes.fabric8.Fabric8Utils.getApplicationNamespace;
 
 /**
  * Kubernetes {@link PropertySourceLocator} for secrets.
diff --git a/spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtilsTests.java b/spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtilsTests.java
index 219d8c7a..d96aaca1 100644
--- a/spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtilsTests.java
+++ b/spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtilsTests.java
@@ -25,22 +25,16 @@ import java.util.Set;
 import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
 import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.fabric8.kubernetes.api.model.SecretBuilder;
-import io.fabric8.kubernetes.client.DefaultKubernetesClient;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
 
 import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.config.MultipleSourcesContainer;
-import org.springframework.cloud.kubernetes.commons.config.NamespaceResolutionFailedException;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties;
 import org.springframework.mock.env.MockEnvironment;
 
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
 /**
  * @author wind57
  */
@@ -49,48 +43,6 @@ class Fabric8ConfigUtilsTests {
 
 	private KubernetesClient client;
 
-	private final DefaultKubernetesClient mockClient = Mockito.mock(DefaultKubernetesClient.class);
-
-	private final KubernetesNamespaceProvider provider = Mockito.mock(KubernetesNamespaceProvider.class);
-
-	@Test
-	void testGetApplicationNamespaceNotPresent() {
-		String result = Fabric8ConfigUtils.getApplicationNamespace(client, "", "target", null);
-		assertThat(result).isEqualTo("test");
-	}
-
-	@Test
-	void testGetApplicationNamespacePresent() {
-		String result = Fabric8ConfigUtils.getApplicationNamespace(client, "namespace", "target", null);
-		assertThat(result).isEqualTo("namespace");
-	}
-
-	@Test
-	void testNamespaceFromNormalizedSource() {
-		String result = Fabric8ConfigUtils.getApplicationNamespace(client, "abc", "target", null);
-		assertThat(result).isEqualTo("abc");
-	}
-
-	@Test
-	void testNamespaceFromProvider() {
-		Mockito.when(provider.getNamespace()).thenReturn("def");
-		String result = Fabric8ConfigUtils.getApplicationNamespace(client, "", "target", provider);
-		assertThat(result).isEqualTo("def");
-	}
-
-	@Test
-	void testNamespaceFromClient() {
-		Mockito.when(mockClient.getNamespace()).thenReturn("qwe");
-		String result = Fabric8ConfigUtils.getApplicationNamespace(mockClient, "", "target", null);
-		assertThat(result).isEqualTo("qwe");
-	}
-
-	@Test
-	void testNamespaceResolutionFailed() {
-		assertThatThrownBy(() -> Fabric8ConfigUtils.getApplicationNamespace(mockClient, "", "target", null))
-				.isInstanceOf(NamespaceResolutionFailedException.class);
-	}
-
 	// secret "my-secret" is deployed without any labels; we search for it by labels
 	// "color=red" and do not find it.
 	@Test
diff --git a/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatch.java b/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatch.java
index 55396232..b0c40a46 100644
--- a/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatch.java
+++ b/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatch.java
@@ -27,8 +27,10 @@ import io.fabric8.kubernetes.client.KubernetesClient;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.cloud.client.discovery.event.HeartbeatEvent;
+import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.discovery.EndpointNameAndNamespace;
 import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
+import org.springframework.cloud.kubernetes.fabric8.Fabric8Utils;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.core.log.LogAccessor;
@@ -45,13 +47,17 @@ public class KubernetesCatalogWatch implements ApplicationEventPublisherAware {
 
 	private final KubernetesDiscoveryProperties properties;
 
+	private final KubernetesNamespaceProvider namespaceProvider;
+
 	private volatile List<EndpointNameAndNamespace> catalogEndpointsState = null;
 
 	private ApplicationEventPublisher publisher;
 
-	public KubernetesCatalogWatch(KubernetesClient kubernetesClient, KubernetesDiscoveryProperties properties) {
+	public KubernetesCatalogWatch(KubernetesClient kubernetesClient, KubernetesDiscoveryProperties properties,
+			KubernetesNamespaceProvider namespaceProvider) {
 		this.kubernetesClient = kubernetesClient;
 		this.properties = properties;
+		this.namespaceProvider = namespaceProvider;
 	}
 
 	@Override
@@ -67,13 +73,28 @@ public class KubernetesCatalogWatch implements ApplicationEventPublisherAware {
 			// endpoints.
 			List<Endpoints> endpoints;
 			if (properties.allNamespaces()) {
+				LOG.debug(() -> "discovering endpoints in all namespaces");
 				endpoints = kubernetesClient.endpoints().inAnyNamespace().withLabels(properties.serviceLabels()).list()
 						.getItems();
 			}
 			else {
-				endpoints = kubernetesClient.endpoints().withLabels(properties.serviceLabels()).list().getItems();
+				String namespace = Fabric8Utils.getApplicationNamespace(kubernetesClient, null, "catalog-watcher",
+						namespaceProvider);
+				LOG.debug(() -> "fabric8 catalog watcher will use namespace : " + namespace);
+				endpoints = kubernetesClient.endpoints().inNamespace(namespace).withLabels(properties.serviceLabels())
+						.list().getItems();
 			}
 
+			/*
+			 * <pre>
+			 *   - An "Endpoints" holds a List of EndpointSubset.
+			 *   - A single EndpointSubset holds a List of EndpointAddress
+			 *
+			 *   - (The union of all EndpointSubsets is the Set of all Endpoints)
+			 *   - Set of Endpoints is the cartesian product of :
+			 *     EndpointSubset::getAddresses and EndpointSubset::getPorts (each is a List)
+			 * </pre>
+			 */
 			List<EndpointNameAndNamespace> currentState = endpoints.stream().map(Endpoints::getSubsets)
 					.filter(Objects::nonNull).flatMap(List::stream).map(EndpointSubset::getAddresses)
 					.filter(Objects::nonNull).flatMap(List::stream).map(EndpointAddress::getTargetRef)
diff --git a/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchAutoConfiguration.java b/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchAutoConfiguration.java
index 83d0e644..c613b39c 100644
--- a/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchAutoConfiguration.java
+++ b/spring-cloud-kubernetes-fabric8-discovery/src/main/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchAutoConfiguration.java
@@ -24,10 +24,12 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.cloud.CloudPlatform;
 import org.springframework.cloud.client.ConditionalOnDiscoveryEnabled;
+import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
 import org.springframework.cloud.kubernetes.fabric8.Fabric8AutoConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
 
 /**
  * Auto configuration for catalog watcher.
@@ -45,8 +47,8 @@ public class KubernetesCatalogWatchAutoConfiguration {
 	@ConditionalOnProperty(name = "spring.cloud.kubernetes.discovery.catalog-services-watch.enabled",
 			matchIfMissing = true)
 	public KubernetesCatalogWatch kubernetesCatalogWatch(KubernetesClient client,
-			KubernetesDiscoveryProperties properties) {
-		return new KubernetesCatalogWatch(client, properties);
+			KubernetesDiscoveryProperties properties, Environment environment) {
+		return new KubernetesCatalogWatch(client, properties, new KubernetesNamespaceProvider(environment));
 	}
 
 }
diff --git a/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/Fabric8KubernetesCatalogWatchTests.java b/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/Fabric8KubernetesCatalogWatchTests.java
new file mode 100644
index 00000000..8536c0d9
--- /dev/null
+++ b/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/Fabric8KubernetesCatalogWatchTests.java
@@ -0,0 +1,282 @@
+/*
+ * Copyright 2013-2022 the original author or authors.
+ *
+ * Licensed 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
+ *
+ *      https://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.springframework.cloud.kubernetes.fabric8.discovery;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import io.fabric8.kubernetes.api.model.EndpointAddress;
+import io.fabric8.kubernetes.api.model.EndpointAddressBuilder;
+import io.fabric8.kubernetes.api.model.EndpointSubset;
+import io.fabric8.kubernetes.api.model.EndpointSubsetBuilder;
+import io.fabric8.kubernetes.api.model.Endpoints;
+import io.fabric8.kubernetes.api.model.EndpointsBuilder;
+import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
+import io.fabric8.kubernetes.api.model.ObjectReferenceBuilder;
+import io.fabric8.kubernetes.client.Config;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+
+import org.springframework.cloud.client.discovery.event.HeartbeatEvent;
+import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
+import org.springframework.cloud.kubernetes.commons.discovery.EndpointNameAndNamespace;
+import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
+import org.springframework.context.ApplicationEventPublisher;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Some tests that use the fabric8 mock client.
+ *
+ * @author wind57
+ */
+@EnableKubernetesMockClient(crud = true, https = false)
+class Fabric8KubernetesCatalogWatchTests {
+
+	private final KubernetesNamespaceProvider namespaceProvider = Mockito.mock(KubernetesNamespaceProvider.class);
+
+	private static final ArgumentCaptor<HeartbeatEvent> HEARTBEAT_EVENT_ARGUMENT_CAPTOR = ArgumentCaptor
+			.forClass(HeartbeatEvent.class);
+
+	private static final ApplicationEventPublisher APPLICATION_EVENT_PUBLISHER = Mockito
+			.mock(ApplicationEventPublisher.class);
+
+	private static KubernetesClient mockClient;
+
+	@BeforeAll
+	static void setUp() {
+		// Configure the kubernetes master url to point to the mock server
+		System.setProperty(Config.KUBERNETES_MASTER_SYSTEM_PROPERTY, mockClient.getConfiguration().getMasterUrl());
+		System.setProperty(Config.KUBERNETES_TRUST_CERT_SYSTEM_PROPERTY, "true");
+		System.setProperty(Config.KUBERNETES_AUTH_TRYKUBECONFIG_SYSTEM_PROPERTY, "false");
+		System.setProperty(Config.KUBERNETES_AUTH_TRYSERVICEACCOUNT_SYSTEM_PROPERTY, "false");
+		System.setProperty(Config.KUBERNETES_NAMESPACE_SYSTEM_PROPERTY, "test");
+		System.setProperty(Config.KUBERNETES_HTTP2_DISABLE, "true");
+	}
+
+	@AfterEach
+	void afterEach() {
+		Mockito.reset(APPLICATION_EVENT_PUBLISHER);
+		mockClient.endpoints().inAnyNamespace().delete();
+	}
+
+	/**
+	 * <pre>
+	 *
+	 *     - we have 5 pods involved in this test
+	 *     - podA in namespaceA with no labels
+	 *     - podB in namespaceA with labels {color=blue}
+	 *     - podC in namespaceA with labels {color=red}
+	 *     - podD in namespaceB with labels {color=blue}
+	 *     - podE in namespaceB with no labels
+	 *
+	 *     We set the namespace to be "namespaceA" and search for labels {color=blue}
+	 *     As a result only one pod is taken: podB
+	 *
+	 * </pre>
+	 */
+	@Test
+	void testEndpointsInSpecificNamespaceWithServiceLabels() {
+
+		KubernetesCatalogWatch watch = createWatcherInSpecificNamespaceAndLabels("namespaceA", Map.of("color", "blue"));
+
+		createSingleEndpoints("namespaceA", Map.of(), "podA");
+		createSingleEndpoints("namespaceA", Map.of("color", "blue"), "podB");
+		createSingleEndpoints("namespaceA", Map.of("color", "red"), "podC");
+		createSingleEndpoints("namespaceB", Map.of("color", "blue"), "podD");
+		createSingleEndpoints("namespaceB", Map.of(), "podE");
+
+		watch.catalogServicesWatch();
+
+		verify(APPLICATION_EVENT_PUBLISHER).publishEvent(HEARTBEAT_EVENT_ARGUMENT_CAPTOR.capture());
+
+		HeartbeatEvent event = HEARTBEAT_EVENT_ARGUMENT_CAPTOR.getValue();
+		assertThat(event.getValue()).isInstanceOf(List.class);
+
+		List<EndpointNameAndNamespace> expectedOutput = List.of(new EndpointNameAndNamespace("podB", "namespaceA"));
+		assertThat(event.getValue()).isEqualTo(expectedOutput);
+	}
+
+	/**
+	 * <pre>
+	 *
+	 *     - we have 5 pods involved in this test
+	 *     - podA in namespaceA with no labels
+	 *     - podB in namespaceA with labels {color=blue}
+	 *     - podC in namespaceA with labels {color=red}
+	 *     - podD in namespaceB with labels {color=blue}
+	 *     - podE in namespaceB with no labels
+	 *
+	 *     We set the namespace to be "namespaceA" and search without labels
+	 *     As a result we get three pods:
+	 *       - podA in namespaceA
+	 *       - podB in namespaceA
+	 *       - pocC in namespaceA
+	 *
+	 * </pre>
+	 */
+	@Test
+	void testEndpointsInSpecificNamespaceWithoutServiceLabels() {
+
+		KubernetesCatalogWatch watch = createWatcherInSpecificNamespaceAndLabels("namespaceA", Map.of());
+
+		createSingleEndpoints("namespaceA", Map.of(), "podA");
+		createSingleEndpoints("namespaceA", Map.of("color", "blue"), "podB");
+		createSingleEndpoints("namespaceA", Map.of("color", "red"), "podC");
+		createSingleEndpoints("namespaceB", Map.of("color", "blue"), "podD");
+		createSingleEndpoints("namespaceB", Map.of(), "podE");
+
+		watch.catalogServicesWatch();
+
+		verify(APPLICATION_EVENT_PUBLISHER).publishEvent(HEARTBEAT_EVENT_ARGUMENT_CAPTOR.capture());
+
+		HeartbeatEvent event = HEARTBEAT_EVENT_ARGUMENT_CAPTOR.getValue();
+		assertThat(event.getValue()).isInstanceOf(List.class);
+
+		List<EndpointNameAndNamespace> expectedOutput = List.of(new EndpointNameAndNamespace("podA", "namespaceA"),
+				new EndpointNameAndNamespace("podB", "namespaceA"), new EndpointNameAndNamespace("podC", "namespaceA"));
+		assertThat(event.getValue()).isEqualTo(expectedOutput);
+	}
+
+	/**
+	 * <pre>
+	 *
+	 *     - we have 5 pods involved in this test
+	 *     - podA in namespaceA with no labels
+	 *     - podB in namespaceA with labels {color=blue}
+	 *     - podC in namespaceA with labels {color=red}
+	 *     - podD in namespaceB with labels {color=blue}
+	 *     - podE in namespaceB with no labels
+	 *
+	 *     We search in all namespaces with labels {color=blue}
+	 *     As a result two pods are taken:
+	 *       - podB in namespaceA
+	 *       - podD in namespaceB
+	 *
+	 * </pre>
+	 */
+	@Test
+	void testEndpointsInAllNamespacesWithServiceLabels() {
+
+		KubernetesCatalogWatch watch = createWatcherInAllNamespacesAndLabels(Map.of("color", "blue"));
+
+		createSingleEndpoints("namespaceA", Map.of(), "podA");
+		createSingleEndpoints("namespaceA", Map.of("color", "blue"), "podB");
+		createSingleEndpoints("namespaceA", Map.of("color", "red"), "podC");
+		createSingleEndpoints("namespaceB", Map.of("color", "blue"), "podD");
+		createSingleEndpoints("namespaceB", Map.of(), "podE");
+
+		watch.catalogServicesWatch();
+
+		verify(APPLICATION_EVENT_PUBLISHER).publishEvent(HEARTBEAT_EVENT_ARGUMENT_CAPTOR.capture());
+
+		HeartbeatEvent event = HEARTBEAT_EVENT_ARGUMENT_CAPTOR.getValue();
+		assertThat(event.getValue()).isInstanceOf(List.class);
+
+		List<EndpointNameAndNamespace> expectedOutput = List.of(new EndpointNameAndNamespace("podB", "namespaceA"),
+				new EndpointNameAndNamespace("podD", "namespaceB"));
+		assertThat(event.getValue()).isEqualTo(expectedOutput);
+	}
+
+	/**
+	 * <pre>
+	 *
+	 *     - we have 5 pods involved in this test
+	 *     - podA in namespaceA with no labels
+	 *     - podB in namespaceA with labels {color=blue}
+	 *     - podC in namespaceA with labels {color=red}
+	 *     - podD in namespaceB with labels {color=blue}
+	 *     - podE in namespaceB with no labels
+	 *
+	 *     We search in all namespaces without labels
+	 *     As a result we get all 5 pods
+	 *
+	 * </pre>
+	 */
+	@Test
+	void testEndpointsInAllNamespacesWithoutServiceLabels() {
+
+		KubernetesCatalogWatch watch = createWatcherInAllNamespacesAndLabels(Map.of());
+
+		createSingleEndpoints("namespaceA", Map.of(), "podA");
+		createSingleEndpoints("namespaceA", Map.of("color", "blue"), "podB");
+		createSingleEndpoints("namespaceA", Map.of("color", "red"), "podC");
+		createSingleEndpoints("namespaceB", Map.of("color", "blue"), "podD");
+		createSingleEndpoints("namespaceB", Map.of(), "podE");
+
+		watch.catalogServicesWatch();
+
+		verify(APPLICATION_EVENT_PUBLISHER).publishEvent(HEARTBEAT_EVENT_ARGUMENT_CAPTOR.capture());
+
+		HeartbeatEvent event = HEARTBEAT_EVENT_ARGUMENT_CAPTOR.getValue();
+		assertThat(event.getValue()).isInstanceOf(List.class);
+
+		List<EndpointNameAndNamespace> expectedOutput = List.of(new EndpointNameAndNamespace("podA", "namespaceA"),
+				new EndpointNameAndNamespace("podB", "namespaceA"), new EndpointNameAndNamespace("podC", "namespaceA"),
+				new EndpointNameAndNamespace("podD", "namespaceB"), new EndpointNameAndNamespace("podE", "namespaceB"));
+		assertThat(event.getValue()).isEqualTo(expectedOutput);
+	}
+
+	private KubernetesCatalogWatch createWatcherInSpecificNamespaceAndLabels(String namespace,
+			Map<String, String> labels) {
+
+		when(namespaceProvider.getNamespace()).thenReturn(namespace);
+
+		// all-namespaces = false
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(true, false, true, 60, false, "",
+				Set.of(), labels, "", null, 0);
+
+		KubernetesCatalogWatch watch = new KubernetesCatalogWatch(mockClient, properties, namespaceProvider);
+		watch.setApplicationEventPublisher(APPLICATION_EVENT_PUBLISHER);
+		return watch;
+
+	}
+
+	private KubernetesCatalogWatch createWatcherInAllNamespacesAndLabels(Map<String, String> labels) {
+
+		// all-namespaces = true
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(true, true, true, 60, false, "",
+				Set.of(), labels, "", null, 0);
+
+		KubernetesCatalogWatch watch = new KubernetesCatalogWatch(mockClient, properties, namespaceProvider);
+		watch.setApplicationEventPublisher(APPLICATION_EVENT_PUBLISHER);
+		return watch;
+
+	}
+
+	private static void createSingleEndpoints(String namespace, Map<String, String> labels, String podName) {
+
+		EndpointAddress endpointAddress = new EndpointAddressBuilder()
+				.withTargetRef(new ObjectReferenceBuilder().withName(podName).withNamespace(namespace).build()).build();
+
+		EndpointSubset endpointSubset = new EndpointSubsetBuilder().withAddresses(List.of(endpointAddress)).build();
+
+		Endpoints endpoints = new EndpointsBuilder()
+				.withMetadata(new ObjectMetaBuilder().withLabels(labels).withName("endpoints-" + podName).build())
+				.withSubsets(List.of(endpointSubset)).build();
+		mockClient.endpoints().inNamespace(namespace).create(endpoints);
+	}
+
+}
diff --git a/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchTest.java b/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchTest.java
index 21307938..63c97206 100644
--- a/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchTest.java
+++ b/spring-cloud-kubernetes-fabric8-discovery/src/test/java/org/springframework/cloud/kubernetes/fabric8/discovery/KubernetesCatalogWatchTest.java
@@ -18,6 +18,8 @@ package org.springframework.cloud.kubernetes.fabric8.discovery;
 
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import io.fabric8.kubernetes.api.model.EndpointAddress;
@@ -26,22 +28,23 @@ import io.fabric8.kubernetes.api.model.Endpoints;
 import io.fabric8.kubernetes.api.model.EndpointsList;
 import io.fabric8.kubernetes.api.model.ObjectReference;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.dsl.FilterWatchListDeletable;
 import io.fabric8.kubernetes.client.dsl.MixedOperation;
+import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.ArgumentCaptor;
 import org.mockito.Mockito;
 
 import org.springframework.cloud.client.discovery.event.HeartbeatEvent;
+import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.discovery.EndpointNameAndNamespace;
 import org.springframework.cloud.kubernetes.commons.discovery.KubernetesDiscoveryProperties;
 import org.springframework.context.ApplicationEventPublisher;
 
 import static java.util.Arrays.stream;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -49,13 +52,14 @@ import static org.mockito.Mockito.when;
 /**
  * @author Oleg Vyukov
  */
-@SuppressWarnings("unchecked")
+@SuppressWarnings({ "unchecked" })
 class KubernetesCatalogWatchTest {
 
 	private static final KubernetesClient CLIENT = Mockito.mock(KubernetesClient.class);
 
-	private final KubernetesCatalogWatch kubernetesCatalogWatch = new KubernetesCatalogWatch(CLIENT,
-			KubernetesDiscoveryProperties.DEFAULT);
+	private final KubernetesNamespaceProvider namespaceProvider = Mockito.mock(KubernetesNamespaceProvider.class);
+
+	private KubernetesCatalogWatch kubernetesCatalogWatch;
 
 	private static final ApplicationEventPublisher APPLICATION_EVENT_PUBLISHER = Mockito
 			.mock(ApplicationEventPublisher.class);
@@ -63,25 +67,29 @@ class KubernetesCatalogWatchTest {
 	private static final MixedOperation<Endpoints, EndpointsList, Resource<Endpoints>> MIXED_OPERATION = Mockito
 			.mock(MixedOperation.class);
 
+	private static final NonNamespaceOperation<Endpoints, EndpointsList, Resource<Endpoints>> NON_NAMESPACE_OPERATION = Mockito
+			.mock(NonNamespaceOperation.class);
+
+	private static final FilterWatchListDeletable<Endpoints, EndpointsList> FILTER_WATCH_LIST_DELETABLE = Mockito
+			.mock(FilterWatchListDeletable.class);
+
 	private static final ArgumentCaptor<HeartbeatEvent> HEARTBEAT_EVENT_ARGUMENT_CAPTOR = ArgumentCaptor
 			.forClass(HeartbeatEvent.class);
 
-	@BeforeEach
-	void setUp() {
-		kubernetesCatalogWatch.setApplicationEventPublisher(APPLICATION_EVENT_PUBLISHER);
-	}
-
 	@AfterEach
 	void afterEach() {
-		Mockito.reset(APPLICATION_EVENT_PUBLISHER, CLIENT, MIXED_OPERATION);
+		Mockito.reset(APPLICATION_EVENT_PUBLISHER, CLIENT, MIXED_OPERATION, FILTER_WATCH_LIST_DELETABLE,
+				NON_NAMESPACE_OPERATION);
 	}
 
 	@Test
 	void testRandomOrderChangePods() {
-		when(MIXED_OPERATION.list()).thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod"))
+
+		createInSpecificNamespaceWatcher();
+
+		when(FILTER_WATCH_LIST_DELETABLE.list())
+				.thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod"))
 				.thenReturn(createSingleEndpointEndpointListByPodName("other-pod", "api-pod"));
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -92,11 +100,11 @@ class KubernetesCatalogWatchTest {
 
 	@Test
 	void testRandomOrderChangePodsAllNamespaces() {
+
+		createInAllNamespaceWatcher();
+
 		when(MIXED_OPERATION.list()).thenReturn(createSingleEndpointEndpointListByPodName("api-pod", "other-pod"))
 				.thenReturn(createSingleEndpointEndpointListByPodName("other-pod", "api-pod"));
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -107,10 +115,12 @@ class KubernetesCatalogWatchTest {
 
 	@Test
 	void testRandomOrderChangeServices() {
-		when(MIXED_OPERATION.list()).thenReturn(createEndpointsListByServiceName("api-service", "other-service"))
+
+		createInSpecificNamespaceWatcher();
+
+		when(FILTER_WATCH_LIST_DELETABLE.list())
+				.thenReturn(createEndpointsListByServiceName("api-service", "other-service"))
 				.thenReturn(createEndpointsListByServiceName("other-service", "api-service"));
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -121,11 +131,11 @@ class KubernetesCatalogWatchTest {
 
 	@Test
 	void testRandomOrderChangeServicesAllNamespaces() {
+
+		createInAllNamespaceWatcher();
+
 		when(MIXED_OPERATION.list()).thenReturn(createEndpointsListByServiceName("api-service", "other-service"))
 				.thenReturn(createEndpointsListByServiceName("other-service", "api-service"));
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -136,10 +146,11 @@ class KubernetesCatalogWatchTest {
 
 	@Test
 	void testEventBody() {
-		when(MIXED_OPERATION.list())
+
+		createInSpecificNamespaceWatcher();
+
+		when(FILTER_WATCH_LIST_DELETABLE.list())
 				.thenReturn(createSingleEndpointListWithNamespace("default", "api-pod", "other-pod"));
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 
@@ -155,11 +166,11 @@ class KubernetesCatalogWatchTest {
 
 	@Test
 	void testEventBodyAllNamespaces() {
+
+		createInAllNamespaceWatcher();
+
 		when(MIXED_OPERATION.list())
 				.thenReturn(createSingleEndpointListWithNamespace("default", "api-pod", "other-pod"));
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 
@@ -176,11 +187,11 @@ class KubernetesCatalogWatchTest {
 	@Test
 	void testEndpointsWithoutSubsets() {
 
+		createInSpecificNamespaceWatcher();
+
 		EndpointsList endpoints = createSingleEndpointEndpointListWithoutSubsets();
 
-		when(MIXED_OPERATION.list()).thenReturn(endpoints);
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
+		when(FILTER_WATCH_LIST_DELETABLE.list()).thenReturn(endpoints);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -192,12 +203,11 @@ class KubernetesCatalogWatchTest {
 	@Test
 	void testEndpointsWithoutSubsetsAllNamespaces() {
 
+		createInAllNamespaceWatcher();
+
 		EndpointsList endpoints = createSingleEndpointEndpointListWithoutSubsets();
 
 		when(MIXED_OPERATION.list()).thenReturn(endpoints);
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -209,12 +219,12 @@ class KubernetesCatalogWatchTest {
 	@Test
 	void testEndpointsWithoutAddresses() {
 
+		createInSpecificNamespaceWatcher();
+
 		EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod");
 		endpoints.getItems().get(0).getSubsets().get(0).setAddresses(null);
 
-		when(MIXED_OPERATION.list()).thenReturn(endpoints);
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
+		when(FILTER_WATCH_LIST_DELETABLE.list()).thenReturn(endpoints);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -226,13 +236,12 @@ class KubernetesCatalogWatchTest {
 	@Test
 	void testEndpointsWithoutAddressesAllNamespaces() {
 
+		createInAllNamespaceWatcher();
+
 		EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod");
 		endpoints.getItems().get(0).getSubsets().get(0).setAddresses(null);
 
 		when(MIXED_OPERATION.list()).thenReturn(endpoints);
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -244,12 +253,12 @@ class KubernetesCatalogWatchTest {
 	@Test
 	void testEndpointsWithoutTargetRefs() {
 
+		createInSpecificNamespaceWatcher();
+
 		EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod");
 		endpoints.getItems().get(0).getSubsets().get(0).getAddresses().get(0).setTargetRef(null);
 
-		when(MIXED_OPERATION.list()).thenReturn(endpoints);
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
+		when(FILTER_WATCH_LIST_DELETABLE.list()).thenReturn(endpoints);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -261,13 +270,12 @@ class KubernetesCatalogWatchTest {
 	@Test
 	void testEndpointsWithoutTargetRefsAllNamespaces() {
 
+		createInAllNamespaceWatcher();
+
 		EndpointsList endpoints = createSingleEndpointEndpointListByPodName("api-pod");
 		endpoints.getItems().get(0).getSubsets().get(0).getAddresses().get(0).setTargetRef(null);
 
 		when(MIXED_OPERATION.list()).thenReturn(endpoints);
-		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace()).thenReturn(MIXED_OPERATION);
-		when(CLIENT.endpoints().inAnyNamespace().withLabels(anyMap())).thenReturn(MIXED_OPERATION);
 
 		kubernetesCatalogWatch.catalogServicesWatch();
 		// second execution on shuffleServices
@@ -336,7 +344,7 @@ class KubernetesCatalogWatchTest {
 			EndpointAddress endpointAddress = new EndpointAddress();
 			endpointAddress.setTargetRef(podRef);
 			return endpointAddress;
-		}).collect(Collectors.toList());
+		}).toList();
 	}
 
 	private List<EndpointAddress> createEndpointAddressWithNamespace(String[] names, String namespace) {
@@ -347,7 +355,36 @@ class KubernetesCatalogWatchTest {
 			EndpointAddress endpointAddress = new EndpointAddress();
 			endpointAddress.setTargetRef(podRef);
 			return endpointAddress;
-		}).collect(Collectors.toList());
+		}).toList();
+	}
+
+	private void createInAllNamespaceWatcher() {
+
+		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
+		when(MIXED_OPERATION.inAnyNamespace()).thenReturn(MIXED_OPERATION);
+		when(MIXED_OPERATION.withLabels(Map.of())).thenReturn(MIXED_OPERATION);
+
+		// all-namespaces = true
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(true, true, true, 60, false, "",
+				Set.of(), Map.of(), "", null, 0);
+
+		kubernetesCatalogWatch = new KubernetesCatalogWatch(CLIENT, properties, namespaceProvider);
+		kubernetesCatalogWatch.setApplicationEventPublisher(APPLICATION_EVENT_PUBLISHER);
+	}
+
+	private void createInSpecificNamespaceWatcher() {
+
+		// all-namespaces = false
+		KubernetesDiscoveryProperties properties = new KubernetesDiscoveryProperties(true, false, true, 60, false, "",
+				Set.of(), Map.of(), "", null, 0);
+
+		kubernetesCatalogWatch = new KubernetesCatalogWatch(CLIENT, properties, namespaceProvider);
+		kubernetesCatalogWatch.setApplicationEventPublisher(APPLICATION_EVENT_PUBLISHER);
+
+		when(namespaceProvider.getNamespace()).thenReturn("catalog-watcher-namespace");
+		when(CLIENT.endpoints()).thenReturn(MIXED_OPERATION);
+		when(MIXED_OPERATION.inNamespace("catalog-watcher-namespace")).thenReturn(NON_NAMESPACE_OPERATION);
+		when(NON_NAMESPACE_OPERATION.withLabels(Map.of())).thenReturn(FILTER_WATCH_LIST_DELETABLE);
 	}
 
 }