GH-2709: Preserve existing JAAS configuration

Resolves #2709

The existing JAAS configuration at the time the
KafkaJaasLoginModuleInitializer's afterSingletonsInstantiated() is now
saved.

The InternalConfiguration class will now search the saved
JAAS Configuration first before returning its own stored
configurationEntries.

This mimics the approach used by Microsoft's SQL Server JDBC driver,
which preserves any existing Configuration while still being able to
return its own default JAAS Configuration.

Added a unit test to KafkaJaasLoginModuleInitializerTests.java to
ensure that an existing configuration can be found.

Author: edanidzerda

spring-kafka/src/main/java/org/springframework/kafka/security/jaas/KafkaJaasLoginModuleInitializer.java

@@ -36,6 +36,7 @@
  *
  * @author Marius Bogoevici
  * @author Gary Russell
+ * @author Edan Idzerda
  *
  * @since 1.3
  */
@@ -134,7 +135,8 @@ public void afterSingletonsInstantiated() {
 					this.options);
 			configurationEntries.put(KAFKA_CLIENT_CONTEXT_NAME,
 					new AppConfigurationEntry[] { kafkaClientConfigurationEntry });
-			Configuration.setConfiguration(new InternalConfiguration(configurationEntries));
+			Configuration.setConfiguration(new InternalConfiguration(configurationEntries,
+					Configuration.getConfiguration()));
 			// Workaround for a 0.9 client issue where even if the Configuration is
 			// set
 			// a system property check is performed.
@@ -157,15 +159,19 @@ private static class InternalConfiguration extends Configuration {
 
 		private final Map<String, AppConfigurationEntry[]> configurationEntries;
 
-		InternalConfiguration(Map<String, AppConfigurationEntry[]> configurationEntries) {
+		private final Configuration delegate;
+
+		InternalConfiguration(Map<String, AppConfigurationEntry[]> configurationEntries, Configuration delegate) {
 			Assert.notNull(configurationEntries, " cannot be null");
 			Assert.notEmpty(configurationEntries, " cannot be empty");
 			this.configurationEntries = configurationEntries;
+			this.delegate = delegate;
 		}
 
 		@Override
 		public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
-			return this.configurationEntries.get(name);
+			AppConfigurationEntry[] conf = this.delegate == null ? null : this.delegate.getAppConfigurationEntry(name);
+			return conf != null ? conf : this.configurationEntries.get(name);
 		}
 
 	}

spring-kafka/src/test/java/org/springframework/kafka/security/jaas/KafkaJaasLoginModuleInitializerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2019 the original author or authors.
+ * Copyright 2017-2023 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.
@@ -27,8 +27,11 @@
 import javax.security.auth.login.AppConfigurationEntry;
 
 import org.apache.kafka.common.security.JaasContext;
+import org.jetbrains.annotations.NotNull;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.io.ClassPathResource;
@@ -40,6 +43,7 @@
 /**
  * @author Marius Bogoevici
  * @author Gary Russell
+ * @author Edan Idzerda
  *
  * @since 1.3
  */
@@ -68,6 +72,16 @@ public void testConfigurationParsedCorrectlyWithKafkaClient() throws Exception {
 		assertThat(appConfigurationEntries.get(0).getOptions()).isEqualTo(kafkaConfigurationArray[0].getOptions());
 	}
 
+	@Test
+	public void testOtherConfigurationFound() {
+		javax.security.auth.login.Configuration configuration = javax.security.auth.login.Configuration
+				.getConfiguration();
+
+		final AppConfigurationEntry[] otherConfiguration = configuration
+				.getAppConfigurationEntry(PreConfiguredJaasConfig.OtherJaasConfigurationName);
+		assertThat(otherConfiguration).hasSize(1);
+	}
+
 	@Configuration
 	public static class Config {
 
@@ -86,4 +100,39 @@ public KafkaJaasLoginModuleInitializer jaasConfig() throws IOException {
 
 	}
 
+	@Configuration
+	public static class PreConfiguredJaasConfig implements BeanPostProcessor {
+
+		private boolean initialized = false;
+
+		public static String OtherJaasConfigurationName = "other-jaas-configuration-name";
+
+		@Override
+		public Object postProcessBeforeInitialization(@NotNull Object bean, @NotNull String beanName)
+				throws BeansException {
+
+			// Install our "other" configuration before the KAFKA_CLIENT_CONTEXT_NAME is installed
+			if (!initialized) {
+				javax.security.auth.login.Configuration.setConfiguration(new OtherJaasConfiguration());
+				initialized = true;
+			}
+			return BeanPostProcessor.super.postProcessBeforeInitialization(bean, beanName);
+		}
+
+		public static class OtherJaasConfiguration extends javax.security.auth.login.Configuration {
+			@Override
+			public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
+				if (name.equals(OtherJaasConfigurationName)) {
+					AppConfigurationEntry dummyAppConfigurationEntry = new AppConfigurationEntry("loginModuleName",
+							AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL, new HashMap<>());
+					return new AppConfigurationEntry[] { dummyAppConfigurationEntry };
+				}
+				else {
+					return null;
+				}
+			}
+		}
+
+	}
+
 }