Add possibility to define the label for watcher 3 (#1063)

* fix

* simplify labels

* fix

* fix test

* trigger

* more simplifications

* very dirty still

* more changes

* more changes

* fix tests

* fix test

* trigger

spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedConfigMapChangeDetector.java

@@ -20,6 +20,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
+import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.informer.ResourceEventHandler;
 import io.kubernetes.client.informer.SharedIndexInformer;
 import io.kubernetes.client.informer.SharedInformerFactory;
@@ -145,7 +146,7 @@ public class KubernetesClientEventBasedConfigMapChangeDetector extends Configura
 		factory.stopAllRegisteredInformers();
 	}
 
-	protected void onEvent(V1ConfigMap configMap) {
+	protected void onEvent(KubernetesObject configMap) {
 		boolean reload = ConfigReloadUtil.reload("config-map", configMap.toString(), propertySourceLocator, environment,
 				KubernetesClientConfigMapPropertySource.class);
 		if (reload) {

spring-cloud-kubernetes-client-config/src/main/java/org/springframework/cloud/kubernetes/client/config/reload/KubernetesClientEventBasedSecretsChangeDetector.java

@@ -20,6 +20,7 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
+import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.informer.ResourceEventHandler;
 import io.kubernetes.client.informer.SharedIndexInformer;
 import io.kubernetes.client.informer.SharedInformerFactory;
@@ -144,7 +145,7 @@ public class KubernetesClientEventBasedSecretsChangeDetector extends Configurati
 		factory.stopAllRegisteredInformers();
 	}
 
-	protected void onEvent(V1Secret secret) {
+	protected void onEvent(KubernetesObject secret) {
 		boolean reload = ConfigReloadUtil.reload("secrets", secret.toString(), propertySourceLocator, environment,
 				KubernetesClientSecretsPropertySource.class);
 		if (reload) {

spring-cloud-kubernetes-commons/src/main/java/org/springframework/cloud/kubernetes/commons/config/reload/ConfigurationUpdateStrategy.java

@@ -24,7 +24,7 @@ import java.util.Objects;
  *
  * @author Nicola Ferraro
  */
-public final record ConfigurationUpdateStrategy(String name, Runnable reloadProcedure) {
+public record ConfigurationUpdateStrategy(String name, Runnable reloadProcedure) {
 
 	public ConfigurationUpdateStrategy(String name, Runnable reloadProcedure) {
 		this.name = Objects.requireNonNull(name, "name cannot be null");

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetector.java

@@ -16,18 +16,14 @@
 
 package org.springframework.cloud.kubernetes.configuration.watcher;
 
+import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1ConfigMap;
 import reactor.core.publisher.Mono;
 
-import org.springframework.cloud.bus.BusProperties;
-import org.springframework.cloud.bus.event.PathDestinationFactory;
-import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent;
 import org.springframework.cloud.kubernetes.client.config.KubernetesClientConfigMapPropertySourceLocator;
 import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigurationUpdateStrategy;
-import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
@@ -35,29 +31,24 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  * @author Ryan Baxter
  * @author Kris Iyer
  */
-public class BusEventBasedConfigMapWatcherChangeDetector extends ConfigMapWatcherChangeDetector {
+final class BusEventBasedConfigMapWatcherChangeDetector extends ConfigMapWatcherChangeDetector {
 
-	private final ApplicationEventPublisher applicationEventPublisher;
+	private final BusRefreshTrigger busRefreshTrigger;
 
-	private final BusProperties busProperties;
-
-	public BusEventBasedConfigMapWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
+	BusEventBasedConfigMapWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
 			ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy,
 			KubernetesClientConfigMapPropertySourceLocator propertySourceLocator,
-			KubernetesNamespaceProvider kubernetesNamespaceProvider, BusProperties busProperties,
+			KubernetesNamespaceProvider kubernetesNamespaceProvider,
 			ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-			ThreadPoolTaskExecutor threadPoolTaskExecutor, ApplicationEventPublisher applicationEventPublisher) {
+			ThreadPoolTaskExecutor threadPoolTaskExecutor, BusRefreshTrigger busRefreshTrigger) {
 		super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider,
 				k8SConfigurationProperties, threadPoolTaskExecutor);
-		this.busProperties = busProperties;
-		this.applicationEventPublisher = applicationEventPublisher;
+		this.busRefreshTrigger = busRefreshTrigger;
 	}
 
 	@Override
-	protected Mono<Void> triggerRefresh(V1ConfigMap configMap) {
-		this.applicationEventPublisher.publishEvent(new RefreshRemoteApplicationEvent(configMap, busProperties.getId(),
-				new PathDestinationFactory().getDestination(configMap.getMetadata().getName())));
-		return Mono.empty();
+	public Mono<Void> triggerRefresh(KubernetesObject configMap) {
+		return busRefreshTrigger.triggerRefresh(configMap);
 	}
 
 }

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetector.java

@@ -16,18 +16,14 @@
 
 package org.springframework.cloud.kubernetes.configuration.watcher;
 
+import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1Secret;
 import reactor.core.publisher.Mono;
 
-import org.springframework.cloud.bus.BusProperties;
-import org.springframework.cloud.bus.event.PathDestinationFactory;
-import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent;
 import org.springframework.cloud.kubernetes.client.config.KubernetesClientSecretsPropertySourceLocator;
 import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigurationUpdateStrategy;
-import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.core.env.ConfigurableEnvironment;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
@@ -35,29 +31,24 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  * @author Ryan Baxter
  * @author Kris Iyer
  */
-public class BusEventBasedSecretsWatcherChangeDetector extends SecretsWatcherChangeDetector {
+final class BusEventBasedSecretsWatcherChangeDetector extends SecretsWatcherChangeDetector {
 
-	private final ApplicationEventPublisher applicationEventPublisher;
+	private final BusRefreshTrigger busRefreshTrigger;
 
-	private final BusProperties busProperties;
-
-	public BusEventBasedSecretsWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
+	BusEventBasedSecretsWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
 			ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy,
 			KubernetesClientSecretsPropertySourceLocator propertySourceLocator,
-			KubernetesNamespaceProvider kubernetesNamespaceProvider, BusProperties busProperties,
+			KubernetesNamespaceProvider kubernetesNamespaceProvider,
 			ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-			ThreadPoolTaskExecutor threadPoolTaskExecutor, ApplicationEventPublisher applicationEventPublisher) {
+			ThreadPoolTaskExecutor threadPoolTaskExecutor, BusRefreshTrigger busRefreshTrigger) {
 		super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider,
 				k8SConfigurationProperties, threadPoolTaskExecutor);
-		this.busProperties = busProperties;
-		this.applicationEventPublisher = applicationEventPublisher;
+		this.busRefreshTrigger = busRefreshTrigger;
 	}
 
 	@Override
-	protected Mono<Void> triggerRefresh(V1Secret secret) {
-		this.applicationEventPublisher.publishEvent(new RefreshRemoteApplicationEvent(secret, busProperties.getId(),
-				new PathDestinationFactory().getDestination(secret.getMetadata().getName())));
-		return Mono.empty();
+	public Mono<Void> triggerRefresh(KubernetesObject secret) {
+		return busRefreshTrigger.triggerRefresh(secret);
 	}
 
 }

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/BusRefreshTrigger.java

@@ -0,0 +1,49 @@
+/*
+ * 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.configuration.watcher;
+
+import io.kubernetes.client.common.KubernetesObject;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.bus.event.PathDestinationFactory;
+import org.springframework.cloud.bus.event.RefreshRemoteApplicationEvent;
+import org.springframework.context.ApplicationEventPublisher;
+
+/**
+ * An event publisher for an 'event bus' type of application.
+ *
+ * @author wind57
+ */
+final class BusRefreshTrigger implements RefreshTrigger {
+
+	private final ApplicationEventPublisher applicationEventPublisher;
+
+	private final String busId;
+
+	BusRefreshTrigger(ApplicationEventPublisher applicationEventPublisher, String busId) {
+		this.applicationEventPublisher = applicationEventPublisher;
+		this.busId = busId;
+	}
+
+	@Override
+	public Mono<Void> triggerRefresh(KubernetesObject configMap) {
+		applicationEventPublisher.publishEvent(new RefreshRemoteApplicationEvent(configMap, busId,
+				new PathDestinationFactory().getDestination(configMap.getMetadata().getName())));
+		return Mono.empty();
+	}
+
+}

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigMapWatcherChangeDetector.java

@@ -18,13 +18,9 @@ package org.springframework.cloud.kubernetes.configuration.watcher;
 
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 
+import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1ConfigMap;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import reactor.core.publisher.Mono;
 
 import org.springframework.cloud.kubernetes.client.config.KubernetesClientConfigMapPropertySourceLocator;
 import org.springframework.cloud.kubernetes.client.config.reload.KubernetesClientEventBasedConfigMapChangeDetector;
@@ -38,60 +34,30 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  * @author Ryan Baxter
  * @author Kris Iyer
  */
-public abstract class ConfigMapWatcherChangeDetector extends KubernetesClientEventBasedConfigMapChangeDetector {
-
-	protected Log log = LogFactory.getLog(getClass());
+abstract class ConfigMapWatcherChangeDetector extends KubernetesClientEventBasedConfigMapChangeDetector
+		implements RefreshTrigger {
 
 	private final ScheduledExecutorService executorService;
 
-	protected ConfigurationWatcherConfigurationProperties k8SConfigurationProperties;
+	private final long refreshDelay;
 
-	public ConfigMapWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
+	ConfigMapWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
 			ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy,
 			KubernetesClientConfigMapPropertySourceLocator propertySourceLocator,
 			KubernetesNamespaceProvider kubernetesNamespaceProvider,
 			ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
 			ThreadPoolTaskExecutor threadPoolTaskExecutor) {
 		super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider);
-
 		this.executorService = Executors.newScheduledThreadPool(k8SConfigurationProperties.getThreadPoolSize(),
 				threadPoolTaskExecutor);
-		this.k8SConfigurationProperties = k8SConfigurationProperties;
+		this.refreshDelay = k8SConfigurationProperties.getRefreshDelay().toMillis();
 	}
 
 	@Override
-	protected void onEvent(V1ConfigMap configMap) {
-		if (isSpringCloudKubernetesConfig(configMap)) {
-			if (log.isDebugEnabled()) {
-				log.debug("Scheduling remote refresh event to be published for ConfigMap "
-						+ configMap.getMetadata().getName() + " to be published in "
-						+ k8SConfigurationProperties.getRefreshDelay().toMillis() + " milliseconds");
-			}
-			executorService.schedule(() -> {
-				try {
-					triggerRefresh(configMap).subscribe();
-				}
-				catch (Throwable t) {
-					log.warn("Error when refreshing ConfigMap " + configMap.getMetadata().getName(), t);
-				}
-			}, k8SConfigurationProperties.getRefreshDelay().toMillis(), TimeUnit.MILLISECONDS);
-		}
-		else {
-			if (log.isDebugEnabled()) {
-				log.debug("Not publishing event. ConfigMap " + configMap.getMetadata().getName()
-						+ " does not contain the label " + ConfigurationWatcherConfigurationProperties.CONFIG_LABEL);
-			}
-		}
-	}
-
-	protected abstract Mono<Void> triggerRefresh(V1ConfigMap configMap);
-
-	private boolean isSpringCloudKubernetesConfig(V1ConfigMap configMap) {
-		if (configMap.getMetadata() == null || configMap.getMetadata().getLabels() == null) {
-			return false;
-		}
-		return Boolean.parseBoolean(configMap.getMetadata().getLabels()
-				.getOrDefault(ConfigurationWatcherConfigurationProperties.CONFIG_LABEL, "false"));
+	protected final void onEvent(KubernetesObject configMap) {
+		// this::refreshTrigger is coming from BusEventBasedConfigMapWatcherChangeDetector
+		WatcherUtil.onEvent(configMap, ConfigurationWatcherConfigurationProperties.CONFIG_LABEL, refreshDelay,
+				executorService, "config-map", this::triggerRefresh);
 	}
 
 }

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherAutoConfiguration.java

@@ -19,6 +19,7 @@ package org.springframework.cloud.kubernetes.configuration.watcher;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
 
 import org.springframework.boot.actuate.autoconfigure.amqp.RabbitHealthContributorAutoConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -45,8 +46,13 @@ import org.springframework.web.reactive.function.client.WebClient;
  */
 @Configuration(proxyBeanMethods = false)
 @EnableConfigurationProperties({ ConfigurationWatcherConfigurationProperties.class })
+@Import({ ConfigurationWatcherAutoConfiguration.RefreshTriggerConfiguration.class })
 public class ConfigurationWatcherAutoConfiguration {
 
+	private static final String AMQP = "bus-amqp";
+
+	private static final String KAFKA = "bus-kafka";
+
 	@Bean
 	@ConditionalOnMissingBean
 	public WebClient webClient(WebClient.Builder webClientBuilder) {
@@ -60,11 +66,11 @@ public class ConfigurationWatcherAutoConfiguration {
 			CoreV1Api coreV1Api, KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator,
 			ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy,
 			ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-			KubernetesNamespaceProvider namespaceProvider, ThreadPoolTaskExecutor threadFactory, WebClient webClient,
-			KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient) {
+			KubernetesNamespaceProvider namespaceProvider, ThreadPoolTaskExecutor threadFactory,
+			HttpRefreshTrigger httpRefreshTrigger) {
 		return new HttpBasedConfigMapWatchChangeDetector(coreV1Api, environment, properties, strategy,
-				configMapPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory, webClient,
-				kubernetesReactiveDiscoveryClient);
+				configMapPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory,
+				httpRefreshTrigger);
 	}
 
 	@Bean
@@ -75,83 +81,99 @@ public class ConfigurationWatcherAutoConfiguration {
 			KubernetesNamespaceProvider namespaceProvider, ConfigReloadProperties properties,
 			ConfigurationUpdateStrategy strategy,
 			ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-			ThreadPoolTaskExecutor threadFactory, WebClient webClient,
-			KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient) {
+			ThreadPoolTaskExecutor threadFactory, HttpRefreshTrigger httpRefreshTrigger) {
 		return new HttpBasedSecretsWatchChangeDetector(coreV1Api, environment, properties, strategy,
-				secretsPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory, webClient,
-				kubernetesReactiveDiscoveryClient);
+				secretsPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory,
+				httpRefreshTrigger);
 	}
 
 	@Configuration
-	@Profile("bus-amqp")
-	@Import({ ContextFunctionCatalogAutoConfiguration.class, RabbitHealthContributorAutoConfiguration.class })
+	@Profile(AMQP)
+	@Import({ ContextFunctionCatalogAutoConfiguration.class, RabbitHealthContributorAutoConfiguration.class,
+			RefreshTriggerConfiguration.class })
 	static class BusRabbitConfiguration {
 
 		@Bean
 		@ConditionalOnMissingBean(ConfigMapWatcherChangeDetector.class)
 		@ConditionalOnBean(KubernetesClientConfigMapPropertySourceLocator.class)
-		public ConfigMapWatcherChangeDetector busConfigMapChangeWatcher(BusProperties busProperties,
-				AbstractEnvironment environment, CoreV1Api coreV1Api,
-				KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator,
+		public ConfigMapWatcherChangeDetector busConfigMapChangeWatcher(AbstractEnvironment environment,
+				CoreV1Api coreV1Api, KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator,
 				KubernetesNamespaceProvider kubernetesNamespaceProvider, ConfigReloadProperties properties,
 				ConfigurationUpdateStrategy strategy,
 				ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-				ThreadPoolTaskExecutor threadFactory, ApplicationEventPublisher applicationEventPublisher) {
+				ThreadPoolTaskExecutor threadFactory, BusRefreshTrigger busRefreshTrigger) {
 			return new BusEventBasedConfigMapWatcherChangeDetector(coreV1Api, environment, properties, strategy,
-					configMapPropertySourceLocator, kubernetesNamespaceProvider, busProperties,
-					k8SConfigurationProperties, threadFactory, applicationEventPublisher);
+					configMapPropertySourceLocator, kubernetesNamespaceProvider, k8SConfigurationProperties,
+					threadFactory, busRefreshTrigger);
 		}
 
 		@Bean
 		@ConditionalOnMissingBean(SecretsWatcherChangeDetector.class)
 		@ConditionalOnBean(KubernetesClientSecretsPropertySourceLocator.class)
-		public SecretsWatcherChangeDetector busSecretsChangeWatcher(BusProperties busProperties,
-				AbstractEnvironment environment, CoreV1Api coreV1Api,
-				KubernetesClientSecretsPropertySourceLocator secretsPropertySourceLocator,
+		public SecretsWatcherChangeDetector busSecretsChangeWatcher(AbstractEnvironment environment,
+				CoreV1Api coreV1Api, KubernetesClientSecretsPropertySourceLocator secretsPropertySourceLocator,
 				ConfigReloadProperties properties, KubernetesNamespaceProvider kubernetesNamespaceProvider,
 				ConfigurationUpdateStrategy strategy,
 				ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-				ThreadPoolTaskExecutor threadFactory, ApplicationEventPublisher applicationEventPublisher) {
+				ThreadPoolTaskExecutor threadFactory, BusRefreshTrigger busRefreshTrigger) {
 			return new BusEventBasedSecretsWatcherChangeDetector(coreV1Api, environment, properties, strategy,
-					secretsPropertySourceLocator, kubernetesNamespaceProvider, busProperties,
-					k8SConfigurationProperties, threadFactory, applicationEventPublisher);
+					secretsPropertySourceLocator, kubernetesNamespaceProvider, k8SConfigurationProperties,
+					threadFactory, busRefreshTrigger);
 		}
 
 	}
 
 	@Configuration
-	@Profile("bus-kafka")
-	@Import({ ContextFunctionCatalogAutoConfiguration.class })
+	@Profile(KAFKA)
+	@Import({ ContextFunctionCatalogAutoConfiguration.class, RefreshTriggerConfiguration.class })
 	static class BusKafkaConfiguration {
 
 		@Bean
 		@ConditionalOnMissingBean(ConfigMapWatcherChangeDetector.class)
 		@ConditionalOnBean(KubernetesClientConfigMapPropertySourceLocator.class)
-		public ConfigMapWatcherChangeDetector busConfigMapChangeWatcher(BusProperties busProperties,
-				AbstractEnvironment environment, CoreV1Api coreV1Api,
-				KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator,
+		public ConfigMapWatcherChangeDetector busConfigMapChangeWatcher(AbstractEnvironment environment,
+				CoreV1Api coreV1Api, KubernetesClientConfigMapPropertySourceLocator configMapPropertySourceLocator,
 				ConfigReloadProperties properties, KubernetesNamespaceProvider namespaceProvider,
 				ConfigurationUpdateStrategy strategy,
 				ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-				ThreadPoolTaskExecutor threadFactory, ApplicationEventPublisher applicationEventPublisher) {
+				ThreadPoolTaskExecutor threadFactory, BusRefreshTrigger busRefreshTrigger) {
 			return new BusEventBasedConfigMapWatcherChangeDetector(coreV1Api, environment, properties, strategy,
-					configMapPropertySourceLocator, namespaceProvider, busProperties, k8SConfigurationProperties,
-					threadFactory, applicationEventPublisher);
+					configMapPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory,
+					busRefreshTrigger);
 		}
 
 		@Bean
 		@ConditionalOnMissingBean(SecretsWatcherChangeDetector.class)
 		@ConditionalOnBean(KubernetesClientSecretsPropertySourceLocator.class)
-		public SecretsWatcherChangeDetector busSecretsChangeWatcher(BusProperties busProperties,
-				AbstractEnvironment environment, CoreV1Api coreV1Api,
-				KubernetesClientSecretsPropertySourceLocator secretsPropertySourceLocator,
+		public SecretsWatcherChangeDetector busSecretsChangeWatcher(AbstractEnvironment environment,
+				CoreV1Api coreV1Api, KubernetesClientSecretsPropertySourceLocator secretsPropertySourceLocator,
 				ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy,
 				ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
 				ThreadPoolTaskExecutor threadFactory, KubernetesNamespaceProvider namespaceProvider,
-				ApplicationEventPublisher applicationEventPublisher) {
+				BusRefreshTrigger busRefreshTrigger) {
 			return new BusEventBasedSecretsWatcherChangeDetector(coreV1Api, environment, properties, strategy,
-					secretsPropertySourceLocator, namespaceProvider, busProperties, k8SConfigurationProperties,
-					threadFactory, applicationEventPublisher);
+					secretsPropertySourceLocator, namespaceProvider, k8SConfigurationProperties, threadFactory,
+					busRefreshTrigger);
+		}
+
+	}
+
+	@AutoConfiguration
+	static class RefreshTriggerConfiguration {
+
+		@Bean
+		@ConditionalOnMissingBean
+		@Profile({ AMQP, KAFKA })
+		public BusRefreshTrigger busRefreshTrigger(ApplicationEventPublisher applicationEventPublisher,
+				BusProperties busProperties) {
+			return new BusRefreshTrigger(applicationEventPublisher, busProperties.getId());
+		}
+
+		@Bean
+		@ConditionalOnMissingBean
+		public HttpRefreshTrigger httpRefreshTrigger(KubernetesInformerReactiveDiscoveryClient client,
+				ConfigurationWatcherConfigurationProperties properties, WebClient webClient) {
+			return new HttpRefreshTrigger(client, properties, webClient);
 		}
 
 	}

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/ConfigurationWatcherConfigurationProperties.java

@@ -38,6 +38,11 @@ public class ConfigurationWatcherConfigurationProperties {
 	 */
 	public static final String SECRET_LABEL = "spring.cloud.kubernetes.secret";
 
+	/**
+	 * Annotation key for actuator port and path.
+	 */
+	public static final String ANNOTATION_KEY = "boot.spring.io/actuator";
+
 	/**
 	 * Amount of time to delay the posting of the event to allow the app volume to update
 	 * data.

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetector.java

@@ -16,125 +16,40 @@
 
 package org.springframework.cloud.kubernetes.configuration.watcher;
 
-import java.net.URI;
-
+import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1ConfigMap;
-import io.kubernetes.client.openapi.models.V1ObjectMeta;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.cloud.kubernetes.client.config.KubernetesClientConfigMapPropertySourceLocator;
-import org.springframework.cloud.kubernetes.client.discovery.reactive.KubernetesInformerReactiveDiscoveryClient;
 import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigurationUpdateStrategy;
 import org.springframework.core.env.ConfigurableEnvironment;
-import org.springframework.http.ResponseEntity;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.util.StringUtils;
-import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.util.UriComponentsBuilder;
 
 /**
  * @author Ryan Baxter
  * @author Kris Iyer
  */
-public class HttpBasedConfigMapWatchChangeDetector extends ConfigMapWatcherChangeDetector {
+final class HttpBasedConfigMapWatchChangeDetector extends ConfigMapWatcherChangeDetector {
 
-	private Log log = LogFactory.getLog(getClass());
+	private final HttpRefreshTrigger httpRefreshTrigger;
 
-	/**
-	 * Annotation key for actuator port and path.
-	 */
-	public static String ANNOTATION_KEY = "boot.spring.io/actuator";
-
-	private WebClient webClient;
-
-	private KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient;
-
-	public HttpBasedConfigMapWatchChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
+	HttpBasedConfigMapWatchChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
 			ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy,
 			KubernetesClientConfigMapPropertySourceLocator propertySourceLocator,
 			KubernetesNamespaceProvider kubernetesNamespaceProvider,
 			ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-			ThreadPoolTaskExecutor threadPoolTaskExecutor, WebClient webClient,
-			KubernetesInformerReactiveDiscoveryClient k8sReactiveDiscoveryClient) {
+			ThreadPoolTaskExecutor threadPoolTaskExecutor, HttpRefreshTrigger httpRefreshTrigger) {
 		super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider,
 				k8SConfigurationProperties, threadPoolTaskExecutor);
-		this.webClient = webClient;
-		this.kubernetesReactiveDiscoveryClient = k8sReactiveDiscoveryClient;
-	}
 
-	private void setActuatorUriFromAnnotation(UriComponentsBuilder actuatorUriBuilder, String metadataUri) {
-		URI annotationUri = URI.create(metadataUri);
-		actuatorUriBuilder.path(annotationUri.getPath() + "/refresh");
-
-		// The URI may not contain a host so if that is the case the port in the URI will
-		// be -1
-		// The authority of the URI will be :<port> for example :9090, we just need the
-		// 9090 in this case
-		if (annotationUri.getPort() < 0) {
-			if (annotationUri.getAuthority() != null) {
-				actuatorUriBuilder.port(annotationUri.getAuthority().replaceFirst(":", ""));
-			}
-		}
-		else {
-			actuatorUriBuilder.port(annotationUri.getPort());
-		}
-	}
-
-	private URI getActuatorUri(ServiceInstance si) {
-		String metadataUri = si.getMetadata().getOrDefault(ANNOTATION_KEY, "");
-		if (log.isDebugEnabled()) {
-			log.debug("Metadata actuator uri is: " + metadataUri);
-		}
-
-		UriComponentsBuilder actuatorUriBuilder = UriComponentsBuilder.newInstance().scheme(si.getScheme())
-				.host(si.getHost());
-
-		if (!StringUtils.isEmpty(metadataUri)) {
-			if (log.isDebugEnabled()) {
-				log.debug("Found actuator URI in service instance metadata");
-			}
-			setActuatorUriFromAnnotation(actuatorUriBuilder, metadataUri);
-		}
-		else {
-			Integer port = k8SConfigurationProperties.getActuatorPort() < 0 ? si.getPort()
-					: k8SConfigurationProperties.getActuatorPort();
-			actuatorUriBuilder = actuatorUriBuilder.path(k8SConfigurationProperties.getActuatorPath() + "/refresh")
-					.port(port);
-		}
-
-		return actuatorUriBuilder.build().toUri();
-	}
-
-	protected Flux<ResponseEntity<Void>> refresh(V1ObjectMeta objectMeta) {
-
-		return kubernetesReactiveDiscoveryClient.getInstances(objectMeta.getName()).flatMap(si -> {
-			URI actuatorUri = getActuatorUri(si);
-			if (log.isDebugEnabled()) {
-				log.debug("Sending refresh request for " + objectMeta.getName() + " to URI " + actuatorUri.toString());
-			}
-			Mono<ResponseEntity<Void>> response = webClient.post().uri(actuatorUri).retrieve().toBodilessEntity()
-					.doOnSuccess(re -> {
-						if (log.isDebugEnabled()) {
-							log.debug("Refresh sent to " + objectMeta.getName() + " at URI address " + actuatorUri
-									+ " returned a " + re.getStatusCode().toString());
-						}
-					}).doOnError(t -> {
-						log.warn("Refresh sent to " + objectMeta.getName() + " failed", t);
-					});
-			return response;
-		});
+		this.httpRefreshTrigger = httpRefreshTrigger;
 	}
 
 	@Override
-	protected Mono<Void> triggerRefresh(V1ConfigMap configMap) {
-		return refresh(configMap.getMetadata()).then();
+	public Mono<Void> triggerRefresh(KubernetesObject configMap) {
+		return httpRefreshTrigger.triggerRefresh(configMap);
 	}
 
 }

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetector.java

@@ -16,121 +16,40 @@
 
 package org.springframework.cloud.kubernetes.configuration.watcher;
 
-import java.net.URI;
-
+import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1ObjectMeta;
-import io.kubernetes.client.openapi.models.V1Secret;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import org.springframework.cloud.client.ServiceInstance;
 import org.springframework.cloud.kubernetes.client.config.KubernetesClientSecretsPropertySourceLocator;
-import org.springframework.cloud.kubernetes.client.discovery.reactive.KubernetesInformerReactiveDiscoveryClient;
 import org.springframework.cloud.kubernetes.commons.KubernetesNamespaceProvider;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigReloadProperties;
 import org.springframework.cloud.kubernetes.commons.config.reload.ConfigurationUpdateStrategy;
 import org.springframework.core.env.ConfigurableEnvironment;
-import org.springframework.http.ResponseEntity;
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
-import org.springframework.util.StringUtils;
-import org.springframework.web.reactive.function.client.WebClient;
-import org.springframework.web.util.UriComponentsBuilder;
 
 /**
  * @author Ryan Baxter
  * @author Kris Iyer
  */
-public class HttpBasedSecretsWatchChangeDetector extends SecretsWatcherChangeDetector {
+final class HttpBasedSecretsWatchChangeDetector extends SecretsWatcherChangeDetector {
 
-	/**
-	 * Annotation key for actuator port and path.
-	 */
-	public static String ANNOTATION_KEY = "boot.spring.io/actuator";
+	private final HttpRefreshTrigger httpRefreshTrigger;
 
-	private WebClient webClient;
-
-	private KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient;
-
-	public HttpBasedSecretsWatchChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
+	HttpBasedSecretsWatchChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
 			ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy,
 			KubernetesClientSecretsPropertySourceLocator propertySourceLocator,
 			KubernetesNamespaceProvider kubernetesNamespaceProvider,
 			ConfigurationWatcherConfigurationProperties k8SConfigurationProperties,
-			ThreadPoolTaskExecutor threadPoolTaskExecutor, WebClient webClient,
-			KubernetesInformerReactiveDiscoveryClient k8sReactiveDiscoveryClient) {
+			ThreadPoolTaskExecutor threadPoolTaskExecutor, HttpRefreshTrigger httpRefreshTrigger) {
 		super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider,
 				k8SConfigurationProperties, threadPoolTaskExecutor);
-		this.webClient = webClient;
-		this.kubernetesReactiveDiscoveryClient = k8sReactiveDiscoveryClient;
+
+		this.httpRefreshTrigger = httpRefreshTrigger;
 	}
 
 	@Override
-	protected Mono<Void> triggerRefresh(V1Secret secret) {
-		return refresh(secret.getMetadata()).then();
-	}
-
-	private void setActuatorUriFromAnnotation(UriComponentsBuilder actuatorUriBuilder, String metadataUri) {
-		URI annotationUri = URI.create(metadataUri);
-		actuatorUriBuilder.path(annotationUri.getPath() + "/refresh");
-
-		// The URI may not contain a host so if that is the case the port in the URI will
-		// be -1
-		// The authority of the URI will be :<port> for example :9090, we just need the
-		// 9090 in this case
-		if (annotationUri.getPort() < 0) {
-			if (annotationUri.getAuthority() != null) {
-				actuatorUriBuilder.port(annotationUri.getAuthority().replaceFirst(":", ""));
-			}
-		}
-		else {
-			actuatorUriBuilder.port(annotationUri.getPort());
-		}
-	}
-
-	private URI getActuatorUri(ServiceInstance si) {
-		String metadataUri = si.getMetadata().getOrDefault(ANNOTATION_KEY, "");
-		if (log.isDebugEnabled()) {
-			log.debug("Metadata actuator uri is: " + metadataUri);
-		}
-
-		UriComponentsBuilder actuatorUriBuilder = UriComponentsBuilder.newInstance().scheme(si.getScheme())
-				.host(si.getHost());
-
-		if (!StringUtils.isEmpty(metadataUri)) {
-			if (log.isDebugEnabled()) {
-				log.debug("Found actuator URI in service instance metadata");
-			}
-			setActuatorUriFromAnnotation(actuatorUriBuilder, metadataUri);
-		}
-		else {
-			Integer port = k8SConfigurationProperties.getActuatorPort() < 0 ? si.getPort()
-					: k8SConfigurationProperties.getActuatorPort();
-			actuatorUriBuilder = actuatorUriBuilder.path(k8SConfigurationProperties.getActuatorPath() + "/refresh")
-					.port(port);
-		}
-
-		return actuatorUriBuilder.build().toUri();
-	}
-
-	protected Flux<ResponseEntity<Void>> refresh(V1ObjectMeta objectMeta) {
-
-		return kubernetesReactiveDiscoveryClient.getInstances(objectMeta.getName()).flatMap(si -> {
-			URI actuatorUri = getActuatorUri(si);
-			if (log.isDebugEnabled()) {
-				log.debug("Sending refresh request for " + objectMeta.getName() + " to URI " + actuatorUri.toString());
-			}
-			Mono<ResponseEntity<Void>> response = webClient.post().uri(actuatorUri).retrieve().toBodilessEntity()
-					.doOnSuccess(re -> {
-						if (log.isDebugEnabled()) {
-							log.debug("Refresh sent to " + objectMeta.getName() + " at URI address " + actuatorUri
-									+ " returned a " + re.getStatusCode().toString());
-						}
-					}).doOnError(t -> {
-						log.warn("Refresh sent to " + objectMeta.getName() + " failed", t);
-					});
-			return response;
-		});
+	public Mono<Void> triggerRefresh(KubernetesObject secret) {
+		return httpRefreshTrigger.triggerRefresh(secret);
 	}
 
 }

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpRefreshTrigger.java

@@ -0,0 +1,117 @@
+/*
+ * 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.configuration.watcher;
+
+import java.net.URI;
+import java.util.function.Consumer;
+
+import io.kubernetes.client.common.KubernetesObject;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.cloud.client.ServiceInstance;
+import org.springframework.cloud.kubernetes.client.config.reload.KubernetesClientEventBasedSecretsChangeDetector;
+import org.springframework.cloud.kubernetes.client.discovery.reactive.KubernetesInformerReactiveDiscoveryClient;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.StringUtils;
+import org.springframework.web.reactive.function.client.WebClient;
+import org.springframework.web.util.UriComponentsBuilder;
+
+/**
+ * @author wind57
+ */
+final class HttpRefreshTrigger implements RefreshTrigger {
+
+	private static final LogAccessor LOG = new LogAccessor(
+			LogFactory.getLog(KubernetesClientEventBasedSecretsChangeDetector.class));
+
+	private final KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient;
+
+	private final ConfigurationWatcherConfigurationProperties k8SConfigurationProperties;
+
+	private final WebClient webClient;
+
+	HttpRefreshTrigger(KubernetesInformerReactiveDiscoveryClient kubernetesReactiveDiscoveryClient,
+			ConfigurationWatcherConfigurationProperties k8SConfigurationProperties, WebClient webClient) {
+		this.kubernetesReactiveDiscoveryClient = kubernetesReactiveDiscoveryClient;
+		this.k8SConfigurationProperties = k8SConfigurationProperties;
+		this.webClient = webClient;
+	}
+
+	@Override
+	public Mono<Void> triggerRefresh(KubernetesObject kubernetesObject) {
+
+		String name = kubernetesObject.getMetadata().getName();
+
+		return kubernetesReactiveDiscoveryClient.getInstances(name).flatMap(si -> {
+			URI actuatorUri = getActuatorUri(si, k8SConfigurationProperties.getActuatorPath(),
+					k8SConfigurationProperties.getActuatorPort());
+			LOG.debug(() -> "Sending refresh request for " + name + " to URI " + actuatorUri);
+			return webClient.post().uri(actuatorUri).retrieve().toBodilessEntity()
+					.doOnSuccess(onSuccess(name, actuatorUri)).doOnError(onError(name));
+		}).then();
+	}
+
+	private Consumer<ResponseEntity<Void>> onSuccess(String name, URI actuatorUri) {
+		return re -> LOG.debug(() -> "Refresh sent to " + name + " at URI address " + actuatorUri + " returned a "
+				+ re.getStatusCode());
+	}
+
+	private Consumer<Throwable> onError(String name) {
+		return t -> LOG.warn(t, () -> "Refresh sent to " + name + " failed");
+	}
+
+	private URI getActuatorUri(ServiceInstance si, String actuatorPath, int actuatorPort) {
+		String metadataUri = si.getMetadata().getOrDefault(ConfigurationWatcherConfigurationProperties.ANNOTATION_KEY,
+				"");
+		LOG.debug(() -> "Metadata actuator uri is: " + metadataUri);
+
+		UriComponentsBuilder actuatorUriBuilder = UriComponentsBuilder.newInstance().scheme(si.getScheme())
+				.host(si.getHost());
+
+		if (StringUtils.hasText(metadataUri)) {
+			LOG.debug(() -> "Found actuator URI in service instance metadata");
+			setActuatorUriFromAnnotation(actuatorUriBuilder, metadataUri);
+		}
+		else {
+			int port = actuatorPort < 0 ? si.getPort() : actuatorPort;
+			actuatorUriBuilder = actuatorUriBuilder.path(actuatorPath + "/refresh").port(port);
+		}
+
+		return actuatorUriBuilder.build().toUri();
+	}
+
+	private void setActuatorUriFromAnnotation(UriComponentsBuilder actuatorUriBuilder, String metadataUri) {
+		URI annotationUri = URI.create(metadataUri);
+		actuatorUriBuilder.path(annotationUri.getPath() + "/refresh");
+
+		// The URI may not contain a host so if that is the case the port in the URI will
+		// be -1. The authority of the URI will be :<port> for example :9090, we just need
+		// the
+		// 9090 in this case
+		if (annotationUri.getPort() < 0) {
+			if (annotationUri.getAuthority() != null) {
+				actuatorUriBuilder.port(annotationUri.getAuthority().replaceFirst(":", ""));
+			}
+		}
+		else {
+			actuatorUriBuilder.port(annotationUri.getPort());
+		}
+	}
+
+}

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/RefreshTrigger.java

@@ -0,0 +1,34 @@
+/*
+ * 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.configuration.watcher;
+
+import io.kubernetes.client.common.KubernetesObject;
+import reactor.core.publisher.Mono;
+
+/**
+ * Defines the refresh trigger contract.
+ *
+ * @author wind57
+ */
+interface RefreshTrigger {
+
+	/**
+	 * @param kubernetesObject either a config-map or secret at the moment.
+	 */
+	Mono<Void> triggerRefresh(KubernetesObject kubernetesObject);
+
+}

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/SecretsWatcherChangeDetector.java

@@ -18,13 +18,9 @@ package org.springframework.cloud.kubernetes.configuration.watcher;
 
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
 
+import io.kubernetes.client.common.KubernetesObject;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1Secret;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import reactor.core.publisher.Mono;
 
 import org.springframework.cloud.kubernetes.client.config.KubernetesClientSecretsPropertySourceLocator;
 import org.springframework.cloud.kubernetes.client.config.reload.KubernetesClientEventBasedSecretsChangeDetector;
@@ -38,15 +34,14 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  * @author Ryan Baxter
  * @author Kris Iyer
  */
-public abstract class SecretsWatcherChangeDetector extends KubernetesClientEventBasedSecretsChangeDetector {
-
-	protected Log log = LogFactory.getLog(getClass());
+abstract class SecretsWatcherChangeDetector extends KubernetesClientEventBasedSecretsChangeDetector
+		implements RefreshTrigger {
 
 	private final ScheduledExecutorService executorService;
 
-	protected ConfigurationWatcherConfigurationProperties k8SConfigurationProperties;
+	private final long refreshDelay;
 
-	public SecretsWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
+	SecretsWatcherChangeDetector(CoreV1Api coreV1Api, ConfigurableEnvironment environment,
 			ConfigReloadProperties properties, ConfigurationUpdateStrategy strategy,
 			KubernetesClientSecretsPropertySourceLocator propertySourceLocator,
 			KubernetesNamespaceProvider kubernetesNamespaceProvider,
@@ -55,42 +50,14 @@ public abstract class SecretsWatcherChangeDetector extends KubernetesClientEvent
 		super(coreV1Api, environment, properties, strategy, propertySourceLocator, kubernetesNamespaceProvider);
 		this.executorService = Executors.newScheduledThreadPool(k8SConfigurationProperties.getThreadPoolSize(),
 				threadPoolTaskExecutor);
-		this.k8SConfigurationProperties = k8SConfigurationProperties;
+		this.refreshDelay = k8SConfigurationProperties.getRefreshDelay().toMillis();
 	}
 
 	@Override
-	protected void onEvent(V1Secret secret) {
-		if (isSpringCloudKubernetesSecret(secret)) {
-			if (log.isDebugEnabled()) {
-				log.debug("Scheduling remote refresh event to be published for Secret " + secret.getMetadata().getName()
-						+ " to be published in " + k8SConfigurationProperties.getRefreshDelay().toMillis()
-						+ " milliseconds");
-			}
-			executorService.schedule(() -> {
-				try {
-					triggerRefresh(secret).subscribe();
-				}
-				catch (Throwable t) {
-					log.warn("Error when refreshing ConfigMap " + secret.getMetadata().getName(), t);
-				}
-			}, k8SConfigurationProperties.getRefreshDelay().toMillis(), TimeUnit.MILLISECONDS);
-		}
-		else {
-			if (log.isDebugEnabled()) {
-				log.debug("Not publishing event. Secret " + secret.getMetadata().getName()
-						+ " does not contain the label " + ConfigurationWatcherConfigurationProperties.SECRET_LABEL);
-			}
-		}
-	}
-
-	protected abstract Mono<Void> triggerRefresh(V1Secret secret);
-
-	private boolean isSpringCloudKubernetesSecret(V1Secret secret) {
-		if (secret.getMetadata() == null || secret.getMetadata().getLabels() == null) {
-			return false;
-		}
-		return Boolean.parseBoolean(secret.getMetadata().getLabels()
-				.getOrDefault(ConfigurationWatcherConfigurationProperties.SECRET_LABEL, "false"));
+	protected final void onEvent(KubernetesObject secret) {
+		// this::refreshTrigger is coming from BusEventBasedSecretsWatcherChangeDetector
+		WatcherUtil.onEvent(secret, ConfigurationWatcherConfigurationProperties.SECRET_LABEL, refreshDelay,
+				executorService, "secret", this::triggerRefresh);
 	}
 
 }

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/main/java/org/springframework/cloud/kubernetes/configuration/watcher/WatcherUtil.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2013-2020 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.configuration.watcher;
+
+import java.util.Collections;
+import java.util.Optional;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+
+import io.kubernetes.client.common.KubernetesObject;
+import org.apache.commons.logging.LogFactory;
+import reactor.core.publisher.Mono;
+
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * A common place where 'onEvent' code delegates to.
+ *
+ * @author wind57
+ */
+final class WatcherUtil {
+
+	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(WatcherUtil.class));
+
+	private WatcherUtil() {
+	}
+
+	static void onEvent(KubernetesObject kubernetesObject, String label, long refreshDelay,
+			ScheduledExecutorService executorService, String type,
+			Function<KubernetesObject, Mono<Void>> triggerRefresh) {
+
+		String name = kubernetesObject.getMetadata().getName();
+		boolean isSpringCloudKubernetes = isSpringCloudKubernetes(kubernetesObject, label);
+
+		if (isSpringCloudKubernetes) {
+
+			LOG.debug(() -> "Scheduling remote refresh event to be published for " + type + ": " + name
+					+ " to be published in " + refreshDelay + " milliseconds");
+			executorService.schedule(() -> {
+				try {
+					triggerRefresh.apply(kubernetesObject).subscribe();
+				}
+				catch (Throwable t) {
+					LOG.warn(t, "Error when refreshing ConfigMap " + name);
+				}
+			}, refreshDelay, TimeUnit.MILLISECONDS);
+		}
+		else {
+			LOG.debug(() -> "Not publishing event." + type + ": + name + does not contain the label " + label);
+		}
+	}
+
+	private static boolean isSpringCloudKubernetes(KubernetesObject kubernetesObject, String label) {
+		if (kubernetesObject.getMetadata() == null) {
+			return false;
+		}
+		return Boolean.parseBoolean(Optional.ofNullable(kubernetesObject.getMetadata().getLabels())
+				.orElse(Collections.emptyMap()).getOrDefault(label, "false"));
+	}
+
+}

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedConfigMapWatcherChangeDetectorTests.java

@@ -80,8 +80,8 @@ class BusEventBasedConfigMapWatcherChangeDetectorTests {
 		busProperties = new BusProperties();
 		changeDetector = new BusEventBasedConfigMapWatcherChangeDetector(coreV1Api, mockEnvironment,
 				configReloadProperties, UPDATE_STRATEGY, configMapPropertySourceLocator,
-				new KubernetesNamespaceProvider(mockEnvironment), busProperties,
-				configurationWatcherConfigurationProperties, threadPoolTaskExecutor, applicationEventPublisher);
+				new KubernetesNamespaceProvider(mockEnvironment), configurationWatcherConfigurationProperties,
+				threadPoolTaskExecutor, new BusRefreshTrigger(applicationEventPublisher, busProperties.getId()));
 	}
 
 	@Test

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/BusEventBasedSecretsWatcherChangeDetectorTests.java

@@ -80,8 +80,8 @@ class BusEventBasedSecretsWatcherChangeDetectorTests {
 		busProperties = new BusProperties();
 		changeDetector = new BusEventBasedSecretsWatcherChangeDetector(coreV1Api, mockEnvironment,
 				configReloadProperties, UPDATE_STRATEGY, secretsPropertySourceLocator,
-				new KubernetesNamespaceProvider(mockEnvironment), busProperties,
-				configurationWatcherConfigurationProperties, threadPoolTaskExecutor, applicationEventPublisher);
+				new KubernetesNamespaceProvider(mockEnvironment), configurationWatcherConfigurationProperties,
+				threadPoolTaskExecutor, new BusRefreshTrigger(applicationEventPublisher, busProperties.getId()));
 	}
 
 	@Test

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedConfigMapWatchChangeDetectorTests.java

@@ -114,8 +114,8 @@ class HttpBasedConfigMapWatchChangeDetectorTests {
 
 		changeDetector = new HttpBasedConfigMapWatchChangeDetector(coreV1Api, mockEnvironment, configReloadProperties,
 				strategy, configMapPropertySourceLocator, new KubernetesNamespaceProvider(mockEnvironment),
-				configurationWatcherConfigurationProperties, threadPoolTaskExecutor, webClient,
-				reactiveDiscoveryClient);
+				configurationWatcherConfigurationProperties, threadPoolTaskExecutor, new HttpRefreshTrigger(
+						reactiveDiscoveryClient, configurationWatcherConfigurationProperties, webClient));
 	}
 
 	@Test
@@ -152,7 +152,8 @@ class HttpBasedConfigMapWatchChangeDetectorTests {
 		int port = WIRE_MOCK_SERVER.port();
 		WireMock.configureFor("localhost", port);
 		Map<String, String> metadata = new HashMap<>();
-		metadata.put(HttpBasedConfigMapWatchChangeDetector.ANNOTATION_KEY, "http://:" + port + "/my/custom/actuator");
+		metadata.put(ConfigurationWatcherConfigurationProperties.ANNOTATION_KEY,
+				"http://:" + port + "/my/custom/actuator");
 		V1EndpointAddress fooEndpointAddress = new V1EndpointAddress();
 		fooEndpointAddress.setIp("127.0.0.1");
 		fooEndpointAddress.setHostname("localhost");

spring-cloud-kubernetes-controllers/spring-cloud-kubernetes-configuration-watcher/src/test/java/org/springframework/cloud/kubernetes/configuration/watcher/HttpBasedSecretsWatchChangeDetectorTests.java

@@ -98,8 +98,8 @@ class HttpBasedSecretsWatchChangeDetectorTests {
 		WebClient webClient = WebClient.builder().build();
 		changeDetector = new HttpBasedSecretsWatchChangeDetector(coreV1Api, mockEnvironment, configReloadProperties,
 				updateStrategy, secretsPropertySourceLocator, new KubernetesNamespaceProvider(mockEnvironment),
-				configurationWatcherConfigurationProperties, threadPoolTaskExecutor, webClient,
-				reactiveDiscoveryClient);
+				configurationWatcherConfigurationProperties, threadPoolTaskExecutor, new HttpRefreshTrigger(
+						reactiveDiscoveryClient, configurationWatcherConfigurationProperties, webClient));
 	}
 
 	@BeforeAll
@@ -148,7 +148,7 @@ class HttpBasedSecretsWatchChangeDetectorTests {
 	void triggerSecretRefreshWithAnnotationActuatorPath() {
 		WireMock.configureFor("localhost", WIRE_MOCK_SERVER.port());
 		Map<String, String> metadata = new HashMap<>();
-		metadata.put(HttpBasedConfigMapWatchChangeDetector.ANNOTATION_KEY,
+		metadata.put(ConfigurationWatcherConfigurationProperties.ANNOTATION_KEY,
 				"http://:" + WIRE_MOCK_SERVER.port() + "/my/custom/actuator");
 		V1EndpointAddress fooEndpointAddress = new V1EndpointAddress();
 		fooEndpointAddress.setIp("127.0.0.1");