fixes 1176 (#1185)

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

@@ -63,7 +63,7 @@ public class KubernetesClientConfigDataLocationResolver extends KubernetesConfig
 					coreV1Api, configMapProperties, namespaceProvider);
 			if (isRetryEnabledForConfigMap(configMapProperties)) {
 				configMapPropertySourceLocator = new ConfigDataRetryableConfigMapPropertySourceLocator(
-						configMapPropertySourceLocator, configMapProperties);
+						configMapPropertySourceLocator, configMapProperties, new KubernetesClientConfigMapsCache());
 			}
 
 			registerSingle(bootstrapContext, ConfigMapPropertySourceLocator.class, configMapPropertySourceLocator,
@@ -75,7 +75,7 @@ public class KubernetesClientConfigDataLocationResolver extends KubernetesConfig
 					coreV1Api, namespaceProvider, secretsProperties);
 			if (isRetryEnabledForSecrets(secretsProperties)) {
 				secretsPropertySourceLocator = new ConfigDataRetryableSecretsPropertySourceLocator(
-						secretsPropertySourceLocator, secretsProperties);
+						secretsPropertySourceLocator, secretsProperties, new KubernetesClientSecretsCache());
 			}
 
 			registerSingle(bootstrapContext, SecretsPropertySourceLocator.class, secretsPropertySourceLocator,

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

@@ -39,7 +39,7 @@ public class KubernetesClientConfigMapPropertySourceLocator extends ConfigMapPro
 
 	public KubernetesClientConfigMapPropertySourceLocator(CoreV1Api coreV1Api, ConfigMapConfigProperties properties,
 			KubernetesNamespaceProvider kubernetesNamespaceProvider) {
-		super(properties);
+		super(properties, new KubernetesClientConfigMapsCache());
 		this.coreV1Api = coreV1Api;
 		this.kubernetesNamespaceProvider = kubernetesNamespaceProvider;
 	}

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

@@ -0,0 +1,82 @@
+/*
+ * 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.client.config;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1ConfigMap;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.kubernetes.commons.config.ConfigMapCache;
+import org.springframework.cloud.kubernetes.commons.config.StrippedSourceContainer;
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * A cache of V1ConfigMap(s) per namespace. Makes sure we read config maps only once from
+ * a namespace.
+ *
+ * @author wind57
+ */
+final class KubernetesClientConfigMapsCache implements ConfigMapCache {
+
+	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(KubernetesClientConfigMapsCache.class));
+
+	/**
+	 * at the moment our loading of config maps is using a single thread, but might change
+	 * in the future, thus a thread safe structure.
+	 */
+	private static final ConcurrentHashMap<String, List<StrippedSourceContainer>> CACHE = new ConcurrentHashMap<>();
+
+	@Override
+	public void discardAll() {
+		CACHE.clear();
+	}
+
+	static List<StrippedSourceContainer> byNamespace(CoreV1Api coreV1Api, String namespace) {
+		boolean[] b = new boolean[1];
+		List<StrippedSourceContainer> result = CACHE.computeIfAbsent(namespace, x -> {
+			try {
+				b[0] = true;
+				return strippedConfigMaps(coreV1Api
+						.listNamespacedConfigMap(namespace, null, null, null, null, null, null, null, null, null, null)
+						.getItems());
+			}
+			catch (ApiException apiException) {
+				throw new RuntimeException(apiException.getResponseBody(), apiException);
+			}
+		});
+
+		if (b[0]) {
+			LOG.debug(() -> "Loaded all config maps in namespace '" + namespace + "'");
+		}
+		else {
+			LOG.debug(() -> "Loaded (from cache) all config maps in namespace '" + namespace + "'");
+		}
+
+		return result;
+	}
+
+	private static List<StrippedSourceContainer> strippedConfigMaps(List<V1ConfigMap> configMaps) {
+		return configMaps.stream().map(configMap -> new StrippedSourceContainer(configMap.getMetadata().getLabels(),
+				configMap.getMetadata().getName(), configMap.getData())).collect(Collectors.toList());
+	}
+
+}

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

@@ -20,12 +20,8 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.stream.Collectors;
 
-import io.kubernetes.client.openapi.ApiException;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
-import io.kubernetes.client.openapi.models.V1ConfigMap;
-import io.kubernetes.client.openapi.models.V1Secret;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -75,16 +71,13 @@ public final class KubernetesClientConfigUtils {
 	 *     6. gather all the names of the secrets (from 4) + data they hold
 	 * </pre>
 	 */
-	static MultipleSourcesContainer secretsDataByLabels(CoreV1Api client, String namespace, Map<String, String> labels,
-			Environment environment, Set<String> profiles) {
-		List<V1Secret> secrets = secretsSearch(client, namespace);
-		if (ConfigUtils.noSources(secrets, namespace)) {
+	static MultipleSourcesContainer secretsDataByLabels(CoreV1Api coreV1Api, String namespace,
+			Map<String, String> labels, Environment environment, Set<String> profiles) {
+		List<StrippedSourceContainer> strippedSecrets = strippedSecrets(coreV1Api, namespace);
+		if (strippedSecrets.isEmpty()) {
 			return MultipleSourcesContainer.empty();
 		}
-
-		List<StrippedSourceContainer> strippedSources = strippedSecrets(secrets);
-		return ConfigUtils.processLabeledData(strippedSources, environment, labels, namespace, profiles, DECODE);
-
+		return ConfigUtils.processLabeledData(strippedSecrets, environment, labels, namespace, profiles, DECODE);
 	}
 
 	/**
@@ -97,16 +90,13 @@ public final class KubernetesClientConfigUtils {
 	 *     6. gather all the names of the config maps (from 4) + data they hold
 	 * </pre>
 	 */
-	static MultipleSourcesContainer configMapsDataByLabels(CoreV1Api client, String namespace,
+	static MultipleSourcesContainer configMapsDataByLabels(CoreV1Api coreV1Api, String namespace,
 			Map<String, String> labels, Environment environment, Set<String> profiles) {
-
-		List<V1ConfigMap> configMaps = configMapsSearch(client, namespace);
-		if (ConfigUtils.noSources(configMaps, namespace)) {
+		List<StrippedSourceContainer> strippedConfigMaps = strippedConfigMaps(coreV1Api, namespace);
+		if (strippedConfigMaps.isEmpty()) {
 			return MultipleSourcesContainer.empty();
 		}
-
-		List<StrippedSourceContainer> strippedSources = strippedConfigMaps(configMaps);
-		return ConfigUtils.processLabeledData(strippedSources, environment, labels, namespace, profiles, DECODE);
+		return ConfigUtils.processLabeledData(strippedConfigMaps, environment, labels, namespace, profiles, DECODE);
 	}
 
 	/**
@@ -117,16 +107,13 @@ public final class KubernetesClientConfigUtils {
 	 *     4. gather all the names of the secrets + decoded data they hold
 	 * </pre>
 	 */
-	static MultipleSourcesContainer secretsDataByName(CoreV1Api client, String namespace,
+	static MultipleSourcesContainer secretsDataByName(CoreV1Api coreV1Api, String namespace,
 			LinkedHashSet<String> sourceNames, Environment environment) {
-		List<V1Secret> secrets = secretsSearch(client, namespace);
-		if (ConfigUtils.noSources(secrets, namespace)) {
+		List<StrippedSourceContainer> strippedSecrets = strippedSecrets(coreV1Api, namespace);
+		if (strippedSecrets.isEmpty()) {
 			return MultipleSourcesContainer.empty();
 		}
-
-		List<StrippedSourceContainer> strippedSources = strippedSecrets(secrets);
-		return ConfigUtils.processNamedData(strippedSources, environment, sourceNames, namespace, DECODE);
-
+		return ConfigUtils.processNamedData(strippedSecrets, environment, sourceNames, namespace, DECODE);
 	}
 
 	/**
@@ -137,52 +124,30 @@ public final class KubernetesClientConfigUtils {
 	 *     4. gather all the names of the config maps + data they hold
 	 * </pre>
 	 */
-	static MultipleSourcesContainer configMapsDataByName(CoreV1Api client, String namespace,
+	static MultipleSourcesContainer configMapsDataByName(CoreV1Api coreV1Api, String namespace,
 			LinkedHashSet<String> sourceNames, Environment environment) {
-		List<V1ConfigMap> configMaps = configMapsSearch(client, namespace);
-		if (ConfigUtils.noSources(configMaps, namespace)) {
+		List<StrippedSourceContainer> strippedConfigMaps = strippedConfigMaps(coreV1Api, namespace);
+		if (strippedConfigMaps.isEmpty()) {
 			return MultipleSourcesContainer.empty();
 		}
-
-		List<StrippedSourceContainer> strippedSources = strippedConfigMaps(configMaps);
-		return ConfigUtils.processNamedData(strippedSources, environment, sourceNames, namespace, DECODE);
-
+		return ConfigUtils.processNamedData(strippedConfigMaps, environment, sourceNames, namespace, DECODE);
 	}
 
-	private static List<V1Secret> secretsSearch(CoreV1Api client, String namespace) {
-		LOG.debug("Loading all secrets in namespace '" + namespace + "'");
-		try {
-			return client.listNamespacedSecret(namespace, null, null, null, null, null, null, null, null, null, null)
-					.getItems();
+	private static List<StrippedSourceContainer> strippedConfigMaps(CoreV1Api coreV1Api, String namespace) {
+		List<StrippedSourceContainer> strippedConfigMaps = KubernetesClientConfigMapsCache.byNamespace(coreV1Api,
+				namespace);
+		if (strippedConfigMaps.isEmpty()) {
+			LOG.debug("No configmaps in namespace '" + namespace + "'");
 		}
-		catch (ApiException apiException) {
-			throw new RuntimeException(apiException.getResponseBody(), apiException);
+		return strippedConfigMaps;
+	}
+
+	private static List<StrippedSourceContainer> strippedSecrets(CoreV1Api coreV1Api, String namespace) {
+		List<StrippedSourceContainer> strippedSecrets = KubernetesClientSecretsCache.byNamespace(coreV1Api, namespace);
+		if (strippedSecrets.isEmpty()) {
+			LOG.debug("No configmaps in namespace '" + namespace + "'");
 		}
-	}
-
-	private static List<V1ConfigMap> configMapsSearch(CoreV1Api client, String namespace) {
-		LOG.debug("Loading all config maps in namespace '" + namespace + "'");
-		try {
-			return client.listNamespacedConfigMap(namespace, null, null, null, null, null, null, null, null, null, null)
-					.getItems();
-		}
-		catch (ApiException apiException) {
-			throw new RuntimeException(apiException.getResponseBody(), apiException);
-		}
-	}
-
-	private static List<StrippedSourceContainer> strippedSecrets(List<V1Secret> secrets) {
-		return secrets.stream().map(secret -> new StrippedSourceContainer(secret.getMetadata().getLabels(),
-				secret.getMetadata().getName(), transform(secret.getData()))).collect(Collectors.toList());
-	}
-
-	private static List<StrippedSourceContainer> strippedConfigMaps(List<V1ConfigMap> configMaps) {
-		return configMaps.stream().map(configMap -> new StrippedSourceContainer(configMap.getMetadata().getLabels(),
-				configMap.getMetadata().getName(), configMap.getData())).collect(Collectors.toList());
-	}
-
-	private static Map<String, String> transform(Map<String, byte[]> in) {
-		return in.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, en -> new String(en.getValue())));
+		return strippedSecrets;
 	}
 
 }

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

@@ -0,0 +1,87 @@
+/*
+ * 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.client.config;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1Secret;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.kubernetes.commons.config.SecretsCache;
+import org.springframework.cloud.kubernetes.commons.config.StrippedSourceContainer;
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * A cache of V1ConfigMap(s) per namespace. Makes sure we read config maps only once from
+ * a namespace.
+ *
+ * @author wind57
+ */
+public class KubernetesClientSecretsCache implements SecretsCache {
+
+	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(KubernetesClientConfigMapsCache.class));
+
+	/**
+	 * at the moment our loading of config maps is using a single thread, but might change
+	 * in the future, thus a thread safe structure.
+	 */
+	private static final ConcurrentHashMap<String, List<StrippedSourceContainer>> CACHE = new ConcurrentHashMap<>();
+
+	@Override
+	public void discardAll() {
+		CACHE.clear();
+	}
+
+	static List<StrippedSourceContainer> byNamespace(CoreV1Api coreV1Api, String namespace) {
+		boolean[] b = new boolean[1];
+		List<StrippedSourceContainer> result = CACHE.computeIfAbsent(namespace, x -> {
+			try {
+				b[0] = true;
+				return strippedSecrets(coreV1Api
+						.listNamespacedSecret(namespace, null, null, null, null, null, null, null, null, null, null)
+						.getItems());
+			}
+			catch (ApiException apiException) {
+				throw new RuntimeException(apiException.getResponseBody(), apiException);
+			}
+		});
+
+		if (b[0]) {
+			LOG.debug(() -> "Loaded all secrets in namespace '" + namespace + "'");
+		}
+		else {
+			LOG.debug(() -> "Loaded (from cache) all secrets in namespace '" + namespace + "'");
+		}
+
+		return result;
+	}
+
+	private static List<StrippedSourceContainer> strippedSecrets(List<V1Secret> secrets) {
+		return secrets.stream().map(secret -> new StrippedSourceContainer(secret.getMetadata().getLabels(),
+				secret.getMetadata().getName(), transform(secret.getData()))).collect(Collectors.toList());
+	}
+
+	private static Map<String, String> transform(Map<String, byte[]> in) {
+		return in.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, en -> new String(en.getValue())));
+	}
+
+}

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

@@ -39,7 +39,7 @@ public class KubernetesClientSecretsPropertySourceLocator extends SecretsPropert
 
 	public KubernetesClientSecretsPropertySourceLocator(CoreV1Api coreV1Api,
 			KubernetesNamespaceProvider kubernetesNamespaceProvider, SecretsConfigProperties secretsConfigProperties) {
-		super(secretsConfigProperties);
+		super(secretsConfigProperties, new KubernetesClientSecretsCache());
 		this.coreV1Api = coreV1Api;
 		this.kubernetesNamespaceProvider = kubernetesNamespaceProvider;
 	}

spring-cloud-kubernetes-client-config/src/test/java/org/springframework/cloud/kubernetes/client/config/KubernetesClientConfigMapPropertySourceTests.java

@@ -93,6 +93,7 @@ class KubernetesClientConfigMapPropertySourceTests {
 	@AfterEach
 	public void afterEach() {
 		WireMock.reset();
+		new KubernetesClientConfigMapsCache().discardAll();
 	}
 
 	@Test

spring-cloud-kubernetes-client-config/src/test/java/org/springframework/cloud/kubernetes/client/config/KubernetesClientSecretsPropertySourceTests.java

@@ -112,6 +112,7 @@ class KubernetesClientSecretsPropertySourceTests {
 	@AfterEach
 	void afterEach() {
 		WireMock.reset();
+		new KubernetesClientSecretsCache().discardAll();
 	}
 
 	@Test

spring-cloud-kubernetes-client-config/src/test/java/org/springframework/cloud/kubernetes/client/config/LabeledConfigMapContextToSourceDataProviderTests.java

@@ -36,7 +36,10 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.LabeledConfigMapNormalizedSource;
 import org.springframework.cloud.kubernetes.commons.config.NormalizedSource;
@@ -51,6 +54,7 @@ import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options
 /**
  * @author wind57
  */
+@ExtendWith(OutputCaptureExtension.class)
 class LabeledConfigMapContextToSourceDataProviderTests {
 
 	private static final Map<String, String> LABELS = new LinkedHashMap<>();
@@ -83,6 +87,7 @@ class LabeledConfigMapContextToSourceDataProviderTests {
 	@AfterEach
 	void afterEach() {
 		WireMock.reset();
+		new KubernetesClientConfigMapsCache().discardAll();
 	}
 
 	/**
@@ -445,6 +450,62 @@ class LabeledConfigMapContextToSourceDataProviderTests {
 
 	}
 
+	/**
+	 * <pre>
+	 *     - one configmap is deployed with label {"color", "red"}
+	 *     - one configmap is deployed with label {"color", "green"}
+	 *
+	 *     - we first search for "red" and find it, and it is retrieved from the cluster via the client.
+	 * 	   - we then search for the "green" one, and it is retrieved from the cache this time.
+	 * </pre>
+	 */
+	@Test
+	void cache(CapturedOutput output) {
+		V1ConfigMap red = new V1ConfigMapBuilder().withMetadata(new V1ObjectMetaBuilder()
+				.withLabels(Map.of("color", "red")).withNamespace(NAMESPACE).withName("red-configmap").build())
+				.addToData("color", "red").build();
+
+		V1ConfigMap green = new V1ConfigMapBuilder().withMetadata(new V1ObjectMetaBuilder()
+				.withLabels(Map.of("color", "green")).withNamespace(NAMESPACE).withName("green-configmap").build())
+				.addToData("color", "green").build();
+
+		V1ConfigMapList configMapList = new V1ConfigMapList().addItemsItem(red).addItemsItem(green);
+
+		stubCall(configMapList);
+		CoreV1Api api = new CoreV1Api();
+
+		NormalizedSource redSource = new LabeledConfigMapNormalizedSource(NAMESPACE, Map.of("color", "red"), false,
+				ConfigUtils.Prefix.DEFAULT, false);
+		KubernetesClientConfigContext redContext = new KubernetesClientConfigContext(api, redSource, NAMESPACE,
+				new MockEnvironment());
+		KubernetesClientContextToSourceData redData = new LabeledConfigMapContextToSourceDataProvider().get();
+		SourceData redSourceData = redData.apply(redContext);
+
+		Assertions.assertEquals(redSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(redSourceData.sourceData().get("color"), "red");
+		Assertions.assertEquals(redSourceData.sourceName(), "configmap.red-configmap.default");
+		Assertions.assertTrue(output.getAll().contains("Loaded all config maps in namespace '" + NAMESPACE + "'"));
+
+		NormalizedSource greenSource = new LabeledConfigMapNormalizedSource(NAMESPACE, Map.of("color", "green"), false,
+				ConfigUtils.Prefix.DEFAULT, false);
+		KubernetesClientConfigContext greenContext = new KubernetesClientConfigContext(api, greenSource, NAMESPACE,
+				new MockEnvironment());
+		KubernetesClientContextToSourceData greenData = new LabeledConfigMapContextToSourceDataProvider().get();
+		SourceData greenSourceData = greenData.apply(greenContext);
+
+		Assertions.assertEquals(greenSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(greenSourceData.sourceData().get("color"), "green");
+		Assertions.assertEquals(greenSourceData.sourceName(), "configmap.green-configmap.default");
+
+		// meaning there is a single entry with such a log statement
+		String[] out = output.getAll().split("Loaded all config maps in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+		// meaning that the second read was done from the cache
+		out = output.getAll().split("Loaded \\(from cache\\) all config maps in namespace");
+		Assertions.assertEquals(out.length, 2);
+	}
+
 	private void stubCall(V1ConfigMapList list) {
 		stubFor(get("/api/v1/namespaces/default/configmaps")
 				.willReturn(aResponse().withStatus(200).withBody(new JSON().serialize(list))));

spring-cloud-kubernetes-client-config/src/test/java/org/springframework/cloud/kubernetes/client/config/LabeledSecretContextToSourceDataProviderTests.java

@@ -37,7 +37,10 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.LabeledSecretNormalizedSource;
 import org.springframework.cloud.kubernetes.commons.config.NormalizedSource;
@@ -52,6 +55,7 @@ import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options
 /**
  * @author wind57
  */
+@ExtendWith(OutputCaptureExtension.class)
 class LabeledSecretContextToSourceDataProviderTests {
 
 	private static final Map<String, String> LABELS = new LinkedHashMap<>();
@@ -80,6 +84,7 @@ class LabeledSecretContextToSourceDataProviderTests {
 	@AfterEach
 	void afterEach() {
 		WireMock.reset();
+		new KubernetesClientSecretsCache().discardAll();
 	}
 
 	/**
@@ -439,6 +444,61 @@ class LabeledSecretContextToSourceDataProviderTests {
 		Assertions.assertEquals(sourceData.sourceName(), "secret.color-secret.default");
 	}
 
+	/**
+	 * <pre>
+	 *     - one secret is deployed with label {"color", "red"}
+	 *     - one secret is deployed with label {"color", "green"}
+	 *
+	 *     - we first search for "red" and find it, and it is retrieved from the cluster via the client.
+	 * 	   - we then search for the "green" one, and it is retrieved from the cache this time.
+	 * </pre>
+	 */
+	@Test
+	void cache(CapturedOutput output) {
+		V1Secret red = new V1SecretBuilder().withMetadata(new V1ObjectMetaBuilder().withLabels(Map.of("color", "red"))
+				.withNamespace(NAMESPACE).withName("red").build()).addToData("color", "red".getBytes()).build();
+
+		V1Secret green = new V1SecretBuilder().withMetadata(new V1ObjectMetaBuilder()
+				.withLabels(Map.of("color", "green")).withNamespace(NAMESPACE).withName("green").build())
+				.addToData("color", "green".getBytes()).build();
+
+		V1SecretList secretList = new V1SecretList().addItemsItem(red).addItemsItem(green);
+
+		stubCall(secretList);
+		CoreV1Api api = new CoreV1Api();
+
+		NormalizedSource redSource = new LabeledSecretNormalizedSource(NAMESPACE, Map.of("color", "red"), false,
+				ConfigUtils.Prefix.DEFAULT, false);
+		KubernetesClientConfigContext redContext = new KubernetesClientConfigContext(api, redSource, NAMESPACE,
+				new MockEnvironment());
+		KubernetesClientContextToSourceData redData = new LabeledSecretContextToSourceDataProvider().get();
+		SourceData redSourceData = redData.apply(redContext);
+
+		Assertions.assertEquals(redSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(redSourceData.sourceData().get("color"), "red");
+		Assertions.assertEquals(redSourceData.sourceName(), "secret.red.default");
+		Assertions.assertTrue(output.getAll().contains("Loaded all secrets in namespace '" + NAMESPACE + "'"));
+
+		NormalizedSource greenSource = new LabeledSecretNormalizedSource(NAMESPACE, Map.of("color", "green"), false,
+				ConfigUtils.Prefix.DEFAULT, false);
+		KubernetesClientConfigContext greenContext = new KubernetesClientConfigContext(api, greenSource, NAMESPACE,
+				new MockEnvironment());
+		KubernetesClientContextToSourceData greenData = new LabeledSecretContextToSourceDataProvider().get();
+		SourceData greenSourceData = greenData.apply(greenContext);
+
+		Assertions.assertEquals(greenSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(greenSourceData.sourceData().get("color"), "green");
+		Assertions.assertEquals(greenSourceData.sourceName(), "secret.green.default");
+
+		// meaning there is a single entry with such a log statement
+		String[] out = output.getAll().split("Loaded all secrets in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+		// meaning that the second read was done from the cache
+		out = output.getAll().split("Loaded \\(from cache\\) all secrets in namespace");
+		Assertions.assertEquals(out.length, 2);
+	}
+
 	private void stubCall(V1SecretList list) {
 		stubFor(get("/api/v1/namespaces/default/secrets")
 				.willReturn(aResponse().withStatus(200).withBody(new JSON().serialize(list))));

spring-cloud-kubernetes-client-config/src/test/java/org/springframework/cloud/kubernetes/client/config/NamedConfigMapContextToSourceDataProviderTests.java

@@ -34,7 +34,10 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.NamedConfigMapNormalizedSource;
 import org.springframework.cloud.kubernetes.commons.config.NormalizedSource;
@@ -49,6 +52,7 @@ import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options
 /**
  * @author wind57
  */
+@ExtendWith(OutputCaptureExtension.class)
 class NamedConfigMapContextToSourceDataProviderTests {
 
 	private static final String NAMESPACE = "default";
@@ -78,6 +82,7 @@ class NamedConfigMapContextToSourceDataProviderTests {
 	@AfterEach
 	void afterEach() {
 		WireMock.reset();
+		new KubernetesClientConfigMapsCache().discardAll();
 	}
 
 	/**
@@ -379,6 +384,61 @@ class NamedConfigMapContextToSourceDataProviderTests {
 		Assertions.assertEquals(sourceData.sourceData(), Map.of("key", "value"));
 	}
 
+	/**
+	 * <pre>
+	 *     - one configmap is deployed with name "red"
+	 *     - one configmap is deployed with name "green"
+	 *
+	 *     - we first search for "red" and find it, and it is retrieved from the cluster via the client.
+	 * 	   - we then search for the "green" one, and it is retrieved from the cache this time.
+	 * </pre>
+	 */
+	@Test
+	void cache(CapturedOutput output) {
+		V1ConfigMap red = new V1ConfigMapBuilder()
+				.withMetadata(new V1ObjectMetaBuilder().withName("red").withNamespace(NAMESPACE).build())
+				.addToData("color", "red").build();
+
+		V1ConfigMap green = new V1ConfigMapBuilder()
+				.withMetadata(new V1ObjectMetaBuilder().withName("green").withNamespace(NAMESPACE).build())
+				.addToData("color", "green").build();
+
+		V1ConfigMapList configMapList = new V1ConfigMapList().addItemsItem(red).addItemsItem(green);
+
+		stubCall(configMapList);
+		CoreV1Api api = new CoreV1Api();
+
+		MockEnvironment environment = new MockEnvironment();
+
+		NormalizedSource redSource = new NamedConfigMapNormalizedSource("red", NAMESPACE, true, false);
+		KubernetesClientConfigContext redContext = new KubernetesClientConfigContext(api, redSource, NAMESPACE,
+				environment);
+		KubernetesClientContextToSourceData redData = new NamedConfigMapContextToSourceDataProvider().get();
+		SourceData redSourceData = redData.apply(redContext);
+
+		Assertions.assertEquals(redSourceData.sourceName(), "configmap.red.default");
+		Assertions.assertEquals(redSourceData.sourceData(), Map.of("color", "red"));
+		Assertions.assertTrue(output.getAll().contains("Loaded all config maps in namespace '" + NAMESPACE + "'"));
+
+		NormalizedSource greenSource = new NamedConfigMapNormalizedSource("green", NAMESPACE, true, true);
+		KubernetesClientConfigContext greenContext = new KubernetesClientConfigContext(api, greenSource, NAMESPACE,
+				environment);
+		KubernetesClientContextToSourceData greenData = new NamedConfigMapContextToSourceDataProvider().get();
+		SourceData greenSourceData = greenData.apply(greenContext);
+
+		Assertions.assertEquals(greenSourceData.sourceName(), "configmap.green.default");
+		Assertions.assertEquals(greenSourceData.sourceData(), Map.of("color", "green"));
+
+		// meaning there is a single entry with such a log statement
+		String[] out = output.getAll().split("Loaded all config maps in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+		// meaning that the second read was done from the cache
+		out = output.getAll().split("Loaded \\(from cache\\) all config maps in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+	}
+
 	private void stubCall(V1ConfigMapList list) {
 		stubFor(get("/api/v1/namespaces/default/configmaps")
 				.willReturn(aResponse().withStatus(200).withBody(new JSON().serialize(list))));

spring-cloud-kubernetes-client-config/src/test/java/org/springframework/cloud/kubernetes/client/config/NamedSecretContextToSourceDataProviderTests.java

@@ -35,7 +35,10 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.NamedSecretNormalizedSource;
 import org.springframework.cloud.kubernetes.commons.config.NormalizedSource;
@@ -47,6 +50,7 @@ import static com.github.tomakehurst.wiremock.client.WireMock.get;
 import static com.github.tomakehurst.wiremock.client.WireMock.stubFor;
 import static com.github.tomakehurst.wiremock.core.WireMockConfiguration.options;
 
+@ExtendWith(OutputCaptureExtension.class)
 class NamedSecretContextToSourceDataProviderTests {
 
 	private static final ConfigUtils.Prefix PREFIX = ConfigUtils.findPrefix("some", false, false, "irrelevant");
@@ -70,6 +74,7 @@ class NamedSecretContextToSourceDataProviderTests {
 	@AfterEach
 	void afterEach() {
 		WireMock.reset();
+		new KubernetesClientSecretsCache().discardAll();
 	}
 
 	/**
@@ -333,6 +338,61 @@ class NamedSecretContextToSourceDataProviderTests {
 		Assertions.assertEquals(sourceData.sourceData(), Map.of("key", "value"));
 	}
 
+	/**
+	 * <pre>
+	 *     - one secret is deployed with name "red"
+	 *     - one secret is deployed with name "green"
+	 *
+	 *     - we first search for "red" and find it, and it is retrieved from the cluster via the client.
+	 * 	   - we then search for the "green" one, and it is retrieved from the cache this time.
+	 * </pre>
+	 */
+	@Test
+	void cache(CapturedOutput output) {
+		V1Secret red = new V1SecretBuilder()
+				.withMetadata(new V1ObjectMetaBuilder().withName("red").withNamespace(NAMESPACE).build())
+				.addToData("color", "red".getBytes()).build();
+
+		V1Secret green = new V1SecretBuilder()
+				.withMetadata(new V1ObjectMetaBuilder().withName("green").withNamespace(NAMESPACE).build())
+				.addToData("color", "green".getBytes()).build();
+
+		V1SecretList configMapList = new V1SecretList().addItemsItem(red).addItemsItem(green);
+
+		stubCall(configMapList);
+		CoreV1Api api = new CoreV1Api();
+
+		MockEnvironment environment = new MockEnvironment();
+
+		NormalizedSource redSource = new NamedSecretNormalizedSource("red", NAMESPACE, true, false);
+		KubernetesClientConfigContext redContext = new KubernetesClientConfigContext(api, redSource, NAMESPACE,
+				environment);
+		KubernetesClientContextToSourceData redData = new NamedSecretContextToSourceDataProvider().get();
+		SourceData redSourceData = redData.apply(redContext);
+
+		Assertions.assertEquals(redSourceData.sourceName(), "secret.red.default");
+		Assertions.assertEquals(redSourceData.sourceData(), Map.of("color", "red"));
+		Assertions.assertTrue(output.getAll().contains("Loaded all secrets in namespace '" + NAMESPACE + "'"));
+
+		NormalizedSource greenSource = new NamedSecretNormalizedSource("green", NAMESPACE, true, true);
+		KubernetesClientConfigContext greenContext = new KubernetesClientConfigContext(api, greenSource, NAMESPACE,
+				environment);
+		KubernetesClientContextToSourceData greenData = new NamedSecretContextToSourceDataProvider().get();
+		SourceData greenSourceData = greenData.apply(greenContext);
+
+		Assertions.assertEquals(greenSourceData.sourceName(), "secret.green.default");
+		Assertions.assertEquals(greenSourceData.sourceData(), Map.of("color", "green"));
+
+		// meaning there is a single entry with such a log statement
+		String[] out = output.getAll().split("Loaded all secrets in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+		// meaning that the second read was done from the cache
+		out = output.getAll().split("Loaded \\(from cache\\) all secrets in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+	}
+
 	private void stubCall(V1SecretList list) {
 		stubFor(get("/api/v1/namespaces/default/secrets")
 				.willReturn(aResponse().withStatus(200).withBody(new JSON().serialize(list))));

spring-cloud-kubernetes-client-config/src/test/resources/logback-test.xml

@@ -0,0 +1,6 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+	<include resource="org/springframework/boot/logging/logback/base.xml"/>
+	<!-- needed for CapturedOutput -->
+	<logger name="org.springframework.cloud.kubernetes.client.config" level="DEBUG"/>
+</configuration>

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

@@ -35,6 +35,11 @@ public class ConfigDataRetryableConfigMapPropertySourceLocator extends ConfigMap
 
 	private ConfigMapPropertySourceLocator configMapPropertySourceLocator;
 
+	/**
+	 * This constructor is deprecated, and we do not use it anymore internally. It will be
+	 * removed in the next major release.
+	 */
+	@Deprecated(forRemoval = true)
 	public ConfigDataRetryableConfigMapPropertySourceLocator(
 			ConfigMapPropertySourceLocator configMapPropertySourceLocator, ConfigMapConfigProperties properties) {
 		super(properties);
@@ -45,6 +50,17 @@ public class ConfigDataRetryableConfigMapPropertySourceLocator extends ConfigMap
 				.build();
 	}
 
+	public ConfigDataRetryableConfigMapPropertySourceLocator(
+			ConfigMapPropertySourceLocator configMapPropertySourceLocator, ConfigMapConfigProperties properties,
+			ConfigMapCache cache) {
+		super(properties, cache);
+		this.configMapPropertySourceLocator = configMapPropertySourceLocator;
+		this.retryTemplate = RetryTemplate.builder().maxAttempts(properties.retry().maxAttempts())
+				.exponentialBackoff(properties.retry().initialInterval(), properties.retry().multiplier(),
+						properties.retry().maxInterval())
+				.build();
+	}
+
 	@Override
 	protected MapPropertySource getMapPropertySource(NormalizedSource normalizedSource,
 			ConfigurableEnvironment environment) {

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

@@ -35,6 +35,11 @@ public class ConfigDataRetryableSecretsPropertySourceLocator extends SecretsProp
 
 	private SecretsPropertySourceLocator secretsPropertySourceLocator;
 
+	/**
+	 * This constructor is deprecated, and we do not use it anymore internally. It will be
+	 * removed in the next major release.
+	 */
+	@Deprecated(forRemoval = true)
 	public ConfigDataRetryableSecretsPropertySourceLocator(SecretsPropertySourceLocator propertySourceLocator,
 			SecretsConfigProperties secretsConfigProperties) {
 		super(secretsConfigProperties);
@@ -45,6 +50,16 @@ public class ConfigDataRetryableSecretsPropertySourceLocator extends SecretsProp
 				.build();
 	}
 
+	public ConfigDataRetryableSecretsPropertySourceLocator(SecretsPropertySourceLocator propertySourceLocator,
+			SecretsConfigProperties secretsConfigProperties, SecretsCache cache) {
+		super(secretsConfigProperties, cache);
+		this.secretsPropertySourceLocator = propertySourceLocator;
+		this.retryTemplate = RetryTemplate.builder().maxAttempts(properties.retry().maxAttempts())
+				.exponentialBackoff(properties.retry().initialInterval(), properties.retry().multiplier(),
+						properties.retry().maxInterval())
+				.build();
+	}
+
 	@Override
 	public PropertySource<?> locate(Environment environment) {
 		return retryTemplate.execute(retryContext -> secretsPropertySourceLocator.locate(environment));

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

@@ -0,0 +1,41 @@
+/*
+ * 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.commons.config;
+
+/**
+ * @author wind57
+ */
+public interface ConfigMapCache {
+
+	/**
+	 * Discards all stored entries from the cache.
+	 */
+	void discardAll();
+
+	/**
+	 * an implementation that does nothing. In the next major release it will become
+	 * absolute and must be removed.
+	 */
+	class NOOPCache implements ConfigMapCache {
+
+		@Override
+		public void discardAll() {
+		}
+
+	}
+
+}

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

@@ -51,10 +51,23 @@ public abstract class ConfigMapPropertySourceLocator implements PropertySourceLo
 
 	private static final Log LOG = LogFactory.getLog(ConfigMapPropertySourceLocator.class);
 
+	private final ConfigMapCache cache;
+
 	protected final ConfigMapConfigProperties properties;
 
+	/**
+	 * This constructor is deprecated, and we do not use it anymore internally. It will be
+	 * removed in the next major release.
+	 */
+	@Deprecated(forRemoval = true)
 	public ConfigMapPropertySourceLocator(ConfigMapConfigProperties properties) {
 		this.properties = properties;
+		this.cache = new ConfigMapCache.NOOPCache();
+	}
+
+	public ConfigMapPropertySourceLocator(ConfigMapConfigProperties properties, ConfigMapCache cache) {
+		this.properties = properties;
+		this.cache = cache;
 	}
 
 	protected abstract MapPropertySource getMapPropertySource(NormalizedSource normalizedSource,
@@ -73,6 +86,7 @@ public abstract class ConfigMapPropertySourceLocator implements PropertySourceLo
 
 			addPropertySourcesFromPaths(environment, composite);
 
+			cache.discardAll();
 			return composite;
 		}
 		return null;

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

@@ -255,14 +255,6 @@ public final class ConfigUtils {
 		return new MultipleSourcesContainer(sourceNames, result);
 	}
 
-	public static boolean noSources(List<?> sources, String namespace) {
-		if (sources == null || sources.isEmpty()) {
-			LOG.debug("No sources in namespace '" + namespace + "'");
-			return true;
-		}
-		return false;
-	}
-
 	private static Map<String, String> decodeData(Map<String, String> data) {
 		Map<String, String> result = new HashMap<>(CollectionUtils.newHashMap(data.size()));
 		data.forEach((key, value) -> result.put(key, new String(Base64.getDecoder().decode(value)).trim()));

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

@@ -0,0 +1,41 @@
+/*
+ * 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.commons.config;
+
+/**
+ * @author wind57
+ */
+public interface SecretsCache {
+
+	/**
+	 * Discards all stored entries from the cache.
+	 */
+	void discardAll();
+
+	/**
+	 * an implementation that does nothing. In the next major release it will become
+	 * absolute and must be removed.
+	 */
+	class NOOPCache implements SecretsCache {
+
+		@Override
+		public void discardAll() {
+		}
+
+	}
+
+}

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

@@ -56,10 +56,23 @@ public abstract class SecretsPropertySourceLocator implements PropertySourceLoca
 
 	private static final Log LOG = LogFactory.getLog(SecretsPropertySourceLocator.class);
 
+	private final SecretsCache cache;
+
 	protected final SecretsConfigProperties properties;
 
+	/**
+	 * This constructor is deprecated, and we do not use it anymore internally. It will be
+	 * removed in the next major release.
+	 */
+	@Deprecated(forRemoval = true)
 	public SecretsPropertySourceLocator(SecretsConfigProperties properties) {
 		this.properties = properties;
+		this.cache = new SecretsCache.NOOPCache();
+	}
+
+	public SecretsPropertySourceLocator(SecretsConfigProperties properties, SecretsCache cache) {
+		this.properties = properties;
+		this.cache = cache;
 	}
 
 	@Override
@@ -77,6 +90,7 @@ public abstract class SecretsPropertySourceLocator implements PropertySourceLoca
 				uniqueSources.forEach(s -> composite.addPropertySource(getMapPropertySourceForSingleSecret(env, s)));
 			}
 
+			cache.discardAll();
 			return composite;
 		}
 		return null;

spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigDataLocationResolver.java

@@ -63,7 +63,7 @@ public class Fabric8ConfigDataLocationResolver extends KubernetesConfigDataLocat
 					kubernetesClient, configMapProperties, namespaceProvider);
 			if (isRetryEnabledForConfigMap(configMapProperties)) {
 				configMapPropertySourceLocator = new ConfigDataRetryableConfigMapPropertySourceLocator(
-						configMapPropertySourceLocator, configMapProperties);
+						configMapPropertySourceLocator, configMapProperties, new Fabric8ConfigMapsCache());
 			}
 
 			registerSingle(bootstrapContext, ConfigMapPropertySourceLocator.class, configMapPropertySourceLocator,
@@ -75,7 +75,7 @@ public class Fabric8ConfigDataLocationResolver extends KubernetesConfigDataLocat
 					kubernetesClient, secretsProperties, namespaceProvider);
 			if (isRetryEnabledForSecrets(secretsProperties)) {
 				secretsPropertySourceLocator = new ConfigDataRetryableSecretsPropertySourceLocator(
-						secretsPropertySourceLocator, secretsProperties);
+						secretsPropertySourceLocator, secretsProperties, new Fabric8SecretsCache());
 			}
 
 			registerSingle(bootstrapContext, SecretsPropertySourceLocator.class, secretsPropertySourceLocator,

spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigMapPropertySourceLocator.java

@@ -45,7 +45,7 @@ public class Fabric8ConfigMapPropertySourceLocator extends ConfigMapPropertySour
 
 	Fabric8ConfigMapPropertySourceLocator(KubernetesClient client, ConfigMapConfigProperties properties,
 			KubernetesNamespaceProvider provider) {
-		super(properties);
+		super(properties, new Fabric8ConfigMapsCache());
 		this.client = client;
 		this.provider = provider;
 	}

spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigMapsCache.java

@@ -0,0 +1,74 @@
+/*
+ * 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.config;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.kubernetes.commons.config.ConfigMapCache;
+import org.springframework.cloud.kubernetes.commons.config.StrippedSourceContainer;
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * A cache of ConfigMaps per namespace. Makes sure we read config maps only once from a
+ * namespace.
+ *
+ * @author wind57
+ */
+final class Fabric8ConfigMapsCache implements ConfigMapCache {
+
+	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(Fabric8ConfigMapsCache.class));
+
+	/**
+	 * at the moment our loading of config maps is using a single thread, but might change
+	 * in the future, thus a thread safe structure.
+	 */
+	private static final ConcurrentHashMap<String, List<StrippedSourceContainer>> CACHE = new ConcurrentHashMap<>();
+
+	@Override
+	public void discardAll() {
+		CACHE.clear();
+	}
+
+	static List<StrippedSourceContainer> byNamespace(KubernetesClient client, String namespace) {
+		boolean[] b = new boolean[1];
+		List<StrippedSourceContainer> result = CACHE.computeIfAbsent(namespace, x -> {
+			b[0] = true;
+			return strippedConfigMaps(client.configMaps().inNamespace(namespace).list().getItems());
+		});
+
+		if (b[0]) {
+			LOG.debug(() -> "Loaded all config maps in namespace '" + namespace + "'");
+		}
+		else {
+			LOG.debug(() -> "Loaded (from cache) all config maps in namespace '" + namespace + "'");
+		}
+
+		return result;
+	}
+
+	private static List<StrippedSourceContainer> strippedConfigMaps(List<ConfigMap> configMaps) {
+		return configMaps.stream().map(configMap -> new StrippedSourceContainer(configMap.getMetadata().getLabels(),
+				configMap.getMetadata().getName(), configMap.getData())).collect(Collectors.toList());
+	}
+
+}

spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtils.java

@@ -20,10 +20,7 @@ import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.stream.Collectors;
 
-import io.fabric8.kubernetes.api.model.ConfigMap;
-import io.fabric8.kubernetes.api.model.Secret;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -73,14 +70,11 @@ public final class Fabric8ConfigUtils {
 	 */
 	static MultipleSourcesContainer secretsDataByLabels(KubernetesClient client, String namespace,
 			Map<String, String> labels, Environment environment, Set<String> profiles) {
-		List<Secret> secrets = secretsSearch(client, namespace);
-		if (ConfigUtils.noSources(secrets, namespace)) {
+		List<StrippedSourceContainer> strippedSecrets = strippedSecrets(client, namespace);
+		if (strippedSecrets.isEmpty()) {
 			return MultipleSourcesContainer.empty();
 		}
-
-		List<StrippedSourceContainer> strippedSources = strippedSecrets(secrets);
-		return ConfigUtils.processLabeledData(strippedSources, environment, labels, namespace, profiles, true);
-
+		return ConfigUtils.processLabeledData(strippedSecrets, environment, labels, namespace, profiles, true);
 	}
 
 	/**
@@ -95,14 +89,12 @@ public final class Fabric8ConfigUtils {
 	 */
 	static MultipleSourcesContainer configMapsDataByLabels(KubernetesClient client, String namespace,
 			Map<String, String> labels, Environment environment, Set<String> profiles) {
-
-		List<ConfigMap> configMaps = configMapsSearch(client, namespace);
-		if (ConfigUtils.noSources(configMaps, namespace)) {
+		List<StrippedSourceContainer> strippedConfigMaps = strippedConfigMaps(client, namespace);
+		if (strippedConfigMaps.isEmpty()) {
 			return MultipleSourcesContainer.empty();
 		}
 
-		List<StrippedSourceContainer> strippedSources = strippedConfigMaps(configMaps);
-		return ConfigUtils.processLabeledData(strippedSources, environment, labels, namespace, profiles, false);
+		return ConfigUtils.processLabeledData(strippedConfigMaps, environment, labels, namespace, profiles, false);
 	}
 
 	/**
@@ -115,14 +107,11 @@ public final class Fabric8ConfigUtils {
 	 */
 	static MultipleSourcesContainer secretsDataByName(KubernetesClient client, String namespace,
 			LinkedHashSet<String> sourceNames, Environment environment) {
-		List<Secret> secrets = secretsSearch(client, namespace);
-		if (ConfigUtils.noSources(secrets, namespace)) {
+		List<StrippedSourceContainer> strippedSecrets = strippedSecrets(client, namespace);
+		if (strippedSecrets.isEmpty()) {
 			return MultipleSourcesContainer.empty();
 		}
-
-		List<StrippedSourceContainer> strippedSources = strippedSecrets(secrets);
-		return ConfigUtils.processNamedData(strippedSources, environment, sourceNames, namespace, true);
-
+		return ConfigUtils.processNamedData(strippedSecrets, environment, sourceNames, namespace, true);
 	}
 
 	/**
@@ -135,36 +124,29 @@ public final class Fabric8ConfigUtils {
 	 */
 	static MultipleSourcesContainer configMapsDataByName(KubernetesClient client, String namespace,
 			LinkedHashSet<String> sourceNames, Environment environment) {
-		List<ConfigMap> configMaps = configMapsSearch(client, namespace);
-		if (ConfigUtils.noSources(configMaps, namespace)) {
+		List<StrippedSourceContainer> strippedConfigMaps = strippedConfigMaps(client, namespace);
+		if (strippedConfigMaps.isEmpty()) {
 			return MultipleSourcesContainer.empty();
 		}
-
-		List<StrippedSourceContainer> strippedSources = strippedConfigMaps(configMaps);
-		return ConfigUtils.processNamedData(strippedSources, environment, sourceNames, namespace, false);
-
+		return ConfigUtils.processNamedData(strippedConfigMaps, environment, sourceNames, namespace, false);
 	}
 
-	// ******** non-exposed methods *******
+	private static List<StrippedSourceContainer> strippedConfigMaps(KubernetesClient client, String namespace) {
+		List<StrippedSourceContainer> strippedConfigMaps = Fabric8ConfigMapsCache.byNamespace(client, namespace);
+		if (strippedConfigMaps.isEmpty()) {
+			LOG.debug("No configmaps in namespace '" + namespace + "'");
+		}
 
-	private static List<Secret> secretsSearch(KubernetesClient client, String namespace) {
-		LOG.debug("Loading all secrets in namespace '" + namespace + "'");
-		return client.secrets().inNamespace(namespace).list().getItems();
+		return strippedConfigMaps;
 	}
 
-	private static List<ConfigMap> configMapsSearch(KubernetesClient client, String namespace) {
-		LOG.debug("Loading all config maps in namespace '" + namespace + "'");
-		return client.configMaps().inNamespace(namespace).list().getItems();
-	}
+	private static List<StrippedSourceContainer> strippedSecrets(KubernetesClient client, String namespace) {
+		List<StrippedSourceContainer> strippedSecrets = Fabric8SecretsCache.byNamespace(client, namespace);
+		if (strippedSecrets.isEmpty()) {
+			LOG.debug("No secrets in namespace '" + namespace + "'");
+		}
 
-	private static List<StrippedSourceContainer> strippedSecrets(List<Secret> secrets) {
-		return secrets.stream().map(secret -> new StrippedSourceContainer(secret.getMetadata().getLabels(),
-				secret.getMetadata().getName(), secret.getData())).collect(Collectors.toList());
-	}
-
-	private static List<StrippedSourceContainer> strippedConfigMaps(List<ConfigMap> configMaps) {
-		return configMaps.stream().map(configMap -> new StrippedSourceContainer(configMap.getMetadata().getLabels(),
-				configMap.getMetadata().getName(), configMap.getData())).collect(Collectors.toList());
+		return strippedSecrets;
 	}
 
 }

spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8SecretsCache.java

@@ -0,0 +1,74 @@
+/*
+ * 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.config;
+
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+
+import io.fabric8.kubernetes.api.model.Secret;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.cloud.kubernetes.commons.config.SecretsCache;
+import org.springframework.cloud.kubernetes.commons.config.StrippedSourceContainer;
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * A cache of ConfigMaps per namespace. Makes sure we read config maps only once from a
+ * namespace.
+ *
+ * @author wind57
+ */
+final class Fabric8SecretsCache implements SecretsCache {
+
+	private static final LogAccessor LOG = new LogAccessor(LogFactory.getLog(Fabric8SecretsCache.class));
+
+	/**
+	 * at the moment our loading of config maps is using a single thread, but might change
+	 * in the future, thus a thread safe structure.
+	 */
+	private static final ConcurrentHashMap<String, List<StrippedSourceContainer>> CACHE = new ConcurrentHashMap<>();
+
+	@Override
+	public void discardAll() {
+		CACHE.clear();
+	}
+
+	static List<StrippedSourceContainer> byNamespace(KubernetesClient client, String namespace) {
+		boolean[] b = new boolean[1];
+		List<StrippedSourceContainer> result = CACHE.computeIfAbsent(namespace, x -> {
+			b[0] = true;
+			return strippedSecrets(client.secrets().inNamespace(namespace).list().getItems());
+		});
+
+		if (b[0]) {
+			LOG.debug(() -> "Loaded all secrets in namespace '" + namespace + "'");
+		}
+		else {
+			LOG.debug(() -> "Loaded (from cache) all secrets in namespace '" + namespace + "'");
+		}
+
+		return result;
+	}
+
+	private static List<StrippedSourceContainer> strippedSecrets(List<Secret> secrets) {
+		return secrets.stream().map(secret -> new StrippedSourceContainer(secret.getMetadata().getLabels(),
+				secret.getMetadata().getName(), secret.getData())).collect(Collectors.toList());
+	}
+
+}

spring-cloud-kubernetes-fabric8-config/src/main/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8SecretsPropertySourceLocator.java

@@ -45,7 +45,7 @@ public class Fabric8SecretsPropertySourceLocator extends SecretsPropertySourceLo
 
 	Fabric8SecretsPropertySourceLocator(KubernetesClient client, SecretsConfigProperties properties,
 			KubernetesNamespaceProvider provider) {
-		super(properties);
+		super(properties, new Fabric8SecretsCache());
 		this.client = client;
 		this.provider = provider;
 	}

spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/ConfigMapsTest.java

@@ -23,6 +23,7 @@ import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
 import io.fabric8.kubernetes.api.model.ConfigMapList;
 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.Test;
 
 import org.springframework.cloud.kubernetes.commons.config.NamedConfigMapNormalizedSource;
@@ -39,6 +40,11 @@ class ConfigMapsTest {
 
 	private static KubernetesClient mockClient;
 
+	@AfterEach
+	void afterEach() {
+		new Fabric8ConfigMapsCache().discardAll();
+	}
+
 	@Test
 	public void testConfigMapList() {
 		mockClient.configMaps().inNamespace("ns1")

spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigMapPropertySourceTests.java

@@ -19,6 +19,7 @@ package org.springframework.cloud.kubernetes.fabric8.config;
 import io.fabric8.kubernetes.client.KubernetesClient;
 import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
 import io.fabric8.kubernetes.client.server.mock.KubernetesMockServer;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
@@ -41,6 +42,11 @@ class Fabric8ConfigMapPropertySourceTests {
 
 	private static final ConfigUtils.Prefix DEFAULT = ConfigUtils.findPrefix("default", false, false, "irrelevant");
 
+	@AfterEach
+	void afterEach() {
+		new Fabric8ConfigMapsCache().discardAll();
+	}
+
 	@Test
 	void constructorShouldThrowExceptionOnFailureWhenFailFastIsEnabled() {
 		String name = "my-config";

spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/Fabric8ConfigUtilsTests.java

@@ -27,6 +27,7 @@ import io.fabric8.kubernetes.api.model.ObjectMetaBuilder;
 import io.fabric8.kubernetes.api.model.SecretBuilder;
 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.Assertions;
 import org.junit.jupiter.api.Test;
 
@@ -43,6 +44,12 @@ class Fabric8ConfigUtilsTests {
 
 	private KubernetesClient client;
 
+	@AfterEach
+	void afterEach() {
+		new Fabric8ConfigMapsCache().discardAll();
+		new Fabric8SecretsCache().discardAll();
+	}
+
 	// secret "my-secret" is deployed without any labels; we search for it by labels
 	// "color=red" and do not find it.
 	@Test

spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/LabeledConfigMapContextToSourceDataProviderTests.java

@@ -30,7 +30,10 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.LabeledConfigMapNormalizedSource;
 import org.springframework.cloud.kubernetes.commons.config.NormalizedSource;
@@ -41,6 +44,7 @@ import org.springframework.mock.env.MockEnvironment;
  * @author wind57
  */
 @EnableKubernetesMockClient(crud = true, https = false)
+@ExtendWith(OutputCaptureExtension.class)
 class LabeledConfigMapContextToSourceDataProviderTests {
 
 	private static final String NAMESPACE = "default";
@@ -76,6 +80,7 @@ class LabeledConfigMapContextToSourceDataProviderTests {
 	@AfterEach
 	void afterEach() {
 		mockClient.configMaps().inNamespace(NAMESPACE).delete();
+		new Fabric8ConfigMapsCache().discardAll();
 	}
 
 	/**
@@ -411,4 +416,56 @@ class LabeledConfigMapContextToSourceDataProviderTests {
 
 	}
 
+	/**
+	 * <pre>
+	 *     - configmap "red-configmap" with label "{color:red}"
+	 *     - configmap "green-configmap" with labels "{color:green}"
+	 *     - we first search for "red" and find it, and it is retrieved from the cluster via the client.
+	 * 	   - we then search for the "green" one, and it is retrieved from the cache this time.
+	 * </pre>
+	 */
+	@Test
+	void cache(CapturedOutput output) {
+		ConfigMap redConfigMap = new ConfigMapBuilder().withNewMetadata().withName("red-configmap")
+				.withLabels(Collections.singletonMap("color", "red")).endMetadata().addToData("one", "1").build();
+
+		ConfigMap greenConfigmap = new ConfigMapBuilder().withNewMetadata().withName("green-configmap")
+				.withLabels(Map.of("color", "green")).endMetadata().addToData("two", "2").build();
+
+		mockClient.configMaps().inNamespace(NAMESPACE).resource(redConfigMap).create();
+		mockClient.configMaps().inNamespace(NAMESPACE).resource(greenConfigmap).create();
+
+		MockEnvironment environment = new MockEnvironment();
+
+		NormalizedSource redNormalizedSource = new LabeledConfigMapNormalizedSource(NAMESPACE,
+				Collections.singletonMap("color", "red"), true, ConfigUtils.Prefix.DELAYED, true);
+		Fabric8ConfigContext redContext = new Fabric8ConfigContext(mockClient, redNormalizedSource, NAMESPACE,
+				environment);
+		Fabric8ContextToSourceData redData = new LabeledConfigMapContextToSourceDataProvider().get();
+		SourceData redSourceData = redData.apply(redContext);
+
+		Assertions.assertEquals(redSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(redSourceData.sourceData().get("red-configmap.one"), "1");
+		Assertions.assertTrue(output.getAll().contains("Loaded all config maps in namespace '" + NAMESPACE + "'"));
+
+		NormalizedSource greenNormalizedSource = new LabeledConfigMapNormalizedSource(NAMESPACE,
+				Collections.singletonMap("color", "green"), true, ConfigUtils.Prefix.DELAYED, true);
+		Fabric8ConfigContext greenContext = new Fabric8ConfigContext(mockClient, greenNormalizedSource, NAMESPACE,
+				environment);
+		Fabric8ContextToSourceData greenData = new LabeledConfigMapContextToSourceDataProvider().get();
+		SourceData greenSourceData = greenData.apply(greenContext);
+
+		Assertions.assertEquals(greenSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(greenSourceData.sourceData().get("green-configmap.two"), "2");
+
+		// meaning there is a single entry with such a log statement
+		String[] out = output.getAll().split("Loaded all config maps in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+		// meaning that the second read was done from the cache
+		out = output.getAll().split("Loaded \\(from cache\\) all config maps in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+	}
+
 }

spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/LabeledSecretContextToSourceDataProviderTests.java

@@ -31,7 +31,10 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.LabeledSecretNormalizedSource;
 import org.springframework.cloud.kubernetes.commons.config.NormalizedSource;
@@ -44,6 +47,7 @@ import org.springframework.mock.env.MockEnvironment;
  * @author wind57
  */
 @EnableKubernetesMockClient(crud = true, https = false)
+@ExtendWith(OutputCaptureExtension.class)
 class LabeledSecretContextToSourceDataProviderTests {
 
 	private static final String NAMESPACE = "default";
@@ -79,6 +83,7 @@ class LabeledSecretContextToSourceDataProviderTests {
 	@AfterEach
 	void afterEach() {
 		mockClient.secrets().inNamespace(NAMESPACE).delete();
+		new Fabric8SecretsCache().discardAll();
 	}
 
 	/**
@@ -446,4 +451,57 @@ class LabeledSecretContextToSourceDataProviderTests {
 		Assertions.assertEquals(sourceData.sourceName(), "secret.color-secret.default");
 	}
 
+	/**
+	 * <pre>
+	 *     - secret "red" with label "{color:red}"
+	 *     - secret "green" with labels "{color:green}"
+	 *     - we first search for "red" and find it, and it is retrieved from the cluster via the client.
+	 * 	   - we then search for the "green" one, and it is retrieved from the cache this time.
+	 * </pre>
+	 */
+	@Test
+	void cache(CapturedOutput output) {
+		Secret red = new SecretBuilder().withNewMetadata().withName("red")
+				.withLabels(Collections.singletonMap("color", "red")).endMetadata()
+				.addToData("one", Base64.getEncoder().encodeToString("1".getBytes())).build();
+
+		Secret green = new SecretBuilder().withNewMetadata().withName("green").withLabels(Map.of("color", "green"))
+				.endMetadata().addToData("two", Base64.getEncoder().encodeToString("2".getBytes())).build();
+
+		mockClient.secrets().inNamespace(NAMESPACE).resource(red).create();
+		mockClient.secrets().inNamespace(NAMESPACE).resource(green).create();
+
+		MockEnvironment environment = new MockEnvironment();
+
+		NormalizedSource redNormalizedSource = new LabeledSecretNormalizedSource(NAMESPACE,
+				Collections.singletonMap("color", "red"), true, ConfigUtils.Prefix.DELAYED, true);
+		Fabric8ConfigContext redContext = new Fabric8ConfigContext(mockClient, redNormalizedSource, NAMESPACE,
+				environment);
+		Fabric8ContextToSourceData redData = new LabeledSecretContextToSourceDataProvider().get();
+		SourceData redSourceData = redData.apply(redContext);
+
+		Assertions.assertEquals(redSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(redSourceData.sourceData().get("red.one"), "1");
+		Assertions.assertTrue(output.getAll().contains("Loaded all secrets in namespace '" + NAMESPACE + "'"));
+
+		NormalizedSource greenNormalizedSource = new LabeledSecretNormalizedSource(NAMESPACE,
+				Collections.singletonMap("color", "green"), true, ConfigUtils.Prefix.DELAYED, true);
+		Fabric8ConfigContext greenContext = new Fabric8ConfigContext(mockClient, greenNormalizedSource, NAMESPACE,
+				environment);
+		Fabric8ContextToSourceData greenData = new LabeledSecretContextToSourceDataProvider().get();
+		SourceData greenSourceData = greenData.apply(greenContext);
+
+		Assertions.assertEquals(greenSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(greenSourceData.sourceData().get("green.two"), "2");
+
+		// meaning there is a single entry with such a log statement
+		String[] out = output.getAll().split("Loaded all secrets in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+		// meaning that the second read was done from the cache
+		out = output.getAll().split("Loaded \\(from cache\\) all secrets in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+	}
+
 }

spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/NamedConfigMapContextToSourceDataProviderTests.java

@@ -28,7 +28,10 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.NamedConfigMapNormalizedSource;
 import org.springframework.cloud.kubernetes.commons.config.NormalizedSource;
@@ -41,6 +44,7 @@ import org.springframework.mock.env.MockEnvironment;
  * @author wind57
  */
 @EnableKubernetesMockClient(crud = true, https = false)
+@ExtendWith(OutputCaptureExtension.class)
 class NamedConfigMapContextToSourceDataProviderTests {
 
 	private static final String NAMESPACE = "default";
@@ -67,6 +71,7 @@ class NamedConfigMapContextToSourceDataProviderTests {
 	@AfterEach
 	void afterEach() {
 		mockClient.configMaps().inNamespace(NAMESPACE).delete();
+		new Fabric8ConfigMapsCache().discardAll();
 	}
 
 	/**
@@ -340,4 +345,55 @@ class NamedConfigMapContextToSourceDataProviderTests {
 		Assertions.assertEquals(sourceData.sourceData(), Collections.singletonMap("key", "value"));
 	}
 
+	/**
+	 * <pre>
+	 *     - two configmaps are deployed : "red", "green", in the same namespace.
+	 *     - we first search for "red" and find it, and it is retrieved from the cluster via the client.
+	 *     - we then search for the "green" one, and it is retrieved from the cache this time.
+	 * </pre>
+	 */
+	@Test
+	void cache(CapturedOutput output) {
+
+		ConfigMap red = new ConfigMapBuilder().withNewMetadata().withName("red").endMetadata()
+				.addToData(COLOR_REALLY_RED).build();
+
+		ConfigMap green = new ConfigMapBuilder().withNewMetadata().withName("green").endMetadata()
+				.addToData("taste", "mango").build();
+
+		mockClient.configMaps().inNamespace(NAMESPACE).resource(red).create();
+		mockClient.configMaps().inNamespace(NAMESPACE).resource(green).create();
+
+		MockEnvironment env = new MockEnvironment();
+		NormalizedSource redNormalizedSource = new NamedConfigMapNormalizedSource("red", NAMESPACE, true, PREFIX,
+				false);
+		Fabric8ConfigContext redContext = new Fabric8ConfigContext(mockClient, redNormalizedSource, NAMESPACE, env);
+		Fabric8ContextToSourceData redData = new NamedConfigMapContextToSourceDataProvider().get();
+		SourceData redSourceData = redData.apply(redContext);
+
+		Assertions.assertEquals(redSourceData.sourceName(), "configmap.red.default");
+		Assertions.assertEquals(redSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(redSourceData.sourceData().get("some.color"), "really-red");
+		Assertions.assertTrue(output.getAll().contains("Loaded all config maps in namespace '" + NAMESPACE + "'"));
+
+		NormalizedSource greenNormalizedSource = new NamedConfigMapNormalizedSource("green", NAMESPACE, true, PREFIX,
+				false);
+		Fabric8ConfigContext greenContext = new Fabric8ConfigContext(mockClient, greenNormalizedSource, NAMESPACE, env);
+		Fabric8ContextToSourceData greenData = new NamedConfigMapContextToSourceDataProvider().get();
+		SourceData greenSourceData = greenData.apply(greenContext);
+
+		Assertions.assertEquals(greenSourceData.sourceName(), "configmap.green.default");
+		Assertions.assertEquals(greenSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(greenSourceData.sourceData().get("some.taste"), "mango");
+
+		// meaning there is a single entry with such a log statement
+		String[] out = output.getAll().split("Loaded all config maps in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+		// meaning that the second read was done from the cache
+		out = output.getAll().split("Loaded \\(from cache\\) all config maps in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+	}
+
 }

spring-cloud-kubernetes-fabric8-config/src/test/java/org/springframework/cloud/kubernetes/fabric8/config/NamedSecretContextToSourceDataProviderTests.java

@@ -29,7 +29,10 @@ import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
 import org.springframework.cloud.kubernetes.commons.config.ConfigUtils;
 import org.springframework.cloud.kubernetes.commons.config.NamedSecretNormalizedSource;
 import org.springframework.cloud.kubernetes.commons.config.NormalizedSource;
@@ -42,6 +45,7 @@ import org.springframework.mock.env.MockEnvironment;
  * @author wind57
  */
 @EnableKubernetesMockClient(crud = true, https = false)
+@ExtendWith(OutputCaptureExtension.class)
 class NamedSecretContextToSourceDataProviderTests {
 
 	private static final String NAMESPACE = "default";
@@ -66,6 +70,7 @@ class NamedSecretContextToSourceDataProviderTests {
 	@AfterEach
 	void afterEach() {
 		mockClient.secrets().inNamespace(NAMESPACE).delete();
+		new Fabric8SecretsCache().discardAll();
 	}
 
 	/**
@@ -306,4 +311,54 @@ class NamedSecretContextToSourceDataProviderTests {
 		Assertions.assertEquals(sourceData.sourceData(), Collections.singletonMap("key", "value"));
 	}
 
+	/**
+	 * <pre>
+	 *     - two secrets are deployed : "red", "green", in the same namespace.
+	 *     - we first search for "red" and find it, and it is retrieved from the cluster via the client.
+	 *     - we then search for the "green" one, and it is retrieved from the cache this time.
+	 * </pre>
+	 */
+	@Test
+	void cache(CapturedOutput output) {
+
+		Secret red = new SecretBuilder().withNewMetadata().withName("red").endMetadata()
+				.addToData("color", Base64.getEncoder().encodeToString("red".getBytes())).build();
+
+		Secret green = new SecretBuilder().withNewMetadata().withName("green").endMetadata()
+				.addToData("taste", Base64.getEncoder().encodeToString("mango".getBytes())).build();
+
+		mockClient.secrets().inNamespace(NAMESPACE).resource(red).create();
+		mockClient.secrets().inNamespace(NAMESPACE).resource(green).create();
+
+		MockEnvironment env = new MockEnvironment();
+		NormalizedSource redNormalizedSource = new NamedSecretNormalizedSource("red", NAMESPACE, true, PREFIX, false);
+		Fabric8ConfigContext redContext = new Fabric8ConfigContext(mockClient, redNormalizedSource, NAMESPACE, env);
+		Fabric8ContextToSourceData redData = new NamedSecretContextToSourceDataProvider().get();
+		SourceData redSourceData = redData.apply(redContext);
+
+		Assertions.assertEquals(redSourceData.sourceName(), "secret.red.default");
+		Assertions.assertEquals(redSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(redSourceData.sourceData().get("some.color"), "red");
+		Assertions.assertTrue(output.getAll().contains("Loaded all secrets in namespace '" + NAMESPACE + "'"));
+
+		NormalizedSource greenNormalizedSource = new NamedSecretNormalizedSource("green", NAMESPACE, true, PREFIX,
+				false);
+		Fabric8ConfigContext greenContext = new Fabric8ConfigContext(mockClient, greenNormalizedSource, NAMESPACE, env);
+		Fabric8ContextToSourceData greenData = new NamedSecretContextToSourceDataProvider().get();
+		SourceData greenSourceData = greenData.apply(greenContext);
+
+		Assertions.assertEquals(greenSourceData.sourceName(), "secret.green.default");
+		Assertions.assertEquals(greenSourceData.sourceData().size(), 1);
+		Assertions.assertEquals(greenSourceData.sourceData().get("some.taste"), "mango");
+
+		// meaning there is a single entry with such a log statement
+		String[] out = output.getAll().split("Loaded all secrets in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+		// meaning that the second read was done from the cache
+		out = output.getAll().split("Loaded \\(from cache\\) all secrets in namespace");
+		Assertions.assertEquals(out.length, 2);
+
+	}
+
 }

spring-cloud-kubernetes-fabric8-config/src/test/resources/logback-test.xml

@@ -4,4 +4,6 @@
 	<logger name="org.hibernate.validator"
 			level="info"/> <!-- Validator prints a lot of debug messages during integration tests -->
 	<logger name="okhttp3.mockwebserver" level="debug"/>
+	<!-- needed for CapturedOutput -->
+	<logger name="org.springframework.cloud.kubernetes.fabric8.config" level="DEBUG"/>
 </configuration>