GH-1092: Fix JAAS config issues (Kafka Streams)

Currently, Kafka Streams binder does not honor JAAS configuration properties.
Address this by adding the same `KafkaJaasLoginModuleInitializer` bean used in
regular Kafka binder.

Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/1092

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java

@@ -42,7 +42,8 @@ import org.springframework.context.annotation.Import;
 @Configuration
 @Import({ KafkaAutoConfiguration.class,
 		MultiBinderPropertiesConfiguration.class,
-		KafkaStreamsBinderHealthIndicatorConfiguration.class})
+		KafkaStreamsBinderHealthIndicatorConfiguration.class,
+		KafkaStreamsJaasConfiguration.class})
 public class GlobalKTableBinderConfiguration {
 
 	@Bean

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java

@@ -40,7 +40,8 @@ import org.springframework.context.annotation.Import;
 @Configuration
 @Import({ KafkaAutoConfiguration.class,
 		MultiBinderPropertiesConfiguration.class,
-		KafkaStreamsBinderHealthIndicatorConfiguration.class})
+		KafkaStreamsBinderHealthIndicatorConfiguration.class,
+		KafkaStreamsJaasConfiguration.class})
 public class KStreamBinderConfiguration {
 
 	@Bean

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java

@@ -42,7 +42,8 @@ import org.springframework.context.annotation.Import;
 @Configuration
 @Import({ KafkaAutoConfiguration.class,
 		MultiBinderPropertiesConfiguration.class,
-		KafkaStreamsBinderHealthIndicatorConfiguration.class})
+		KafkaStreamsBinderHealthIndicatorConfiguration.class,
+		KafkaStreamsJaasConfiguration.class})
 public class KTableBinderConfiguration {
 
 	@Bean

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsJaasConfiguration.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2019-2021 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.stream.binder.kafka.streams;
+
+import java.io.IOException;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.cloud.stream.binder.kafka.properties.JaasLoginModuleConfiguration;
+import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;
+
+/**
+ * Jaas configuration bean for Kafka Streams binder types.
+ *
+ * @author Soby Chacko
+ * @since 3.1.4
+ */
+@Configuration
+public class KafkaStreamsJaasConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean(KafkaJaasLoginModuleInitializer.class)
+	public KafkaJaasLoginModuleInitializer jaasInitializer(
+			KafkaBinderConfigurationProperties configurationProperties)
+			throws IOException {
+		KafkaJaasLoginModuleInitializer kafkaJaasLoginModuleInitializer = new KafkaJaasLoginModuleInitializer();
+		JaasLoginModuleConfiguration jaas = configurationProperties.getJaas();
+		if (jaas != null) {
+			kafkaJaasLoginModuleInitializer.setLoginModule(jaas.getLoginModule());
+
+			KafkaJaasLoginModuleInitializer.ControlFlag controlFlag = jaas
+					.getControlFlag();
+
+			if (controlFlag != null) {
+				kafkaJaasLoginModuleInitializer.setControlFlag(controlFlag);
+			}
+			kafkaJaasLoginModuleInitializer.setOptions(jaas.getOptions());
+		}
+		return kafkaJaasLoginModuleInitializer;
+	}
+}

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2021 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.
@@ -16,6 +16,8 @@
 
 package org.springframework.cloud.stream.binder.kafka.streams.bootstrap;
 
+import javax.security.auth.login.AppConfigurationEntry;
+
 import org.apache.kafka.streams.kstream.GlobalKTable;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
@@ -31,6 +33,8 @@ import org.springframework.cloud.stream.annotation.StreamListener;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.kafka.test.rule.EmbeddedKafkaRule;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /**
  * @author Soby Chacko
  */
@@ -87,6 +91,33 @@ public class KafkaStreamsBinderBootstrapTest {
 		applicationContext.close();
 	}
 
+	@Test
+	public void testKafkaStreamsBinderJaasInitialization() {
+		ConfigurableApplicationContext applicationContext = new SpringApplicationBuilder(
+				SimpleKafkaStreamsApplication.class).web(WebApplicationType.NONE).run(
+				"--spring.cloud.stream.kafka.streams.bindings.input-1.consumer.application-id"
+						+ "=testKafkaStreamsBinderWithStandardConfigurationCanStart",
+				"--spring.cloud.stream.kafka.streams.bindings.input-2.consumer.application-id"
+						+ "=testKafkaStreamsBinderWithStandardConfigurationCanStart-foo",
+				"--spring.cloud.stream.kafka.streams.bindings.input-3.consumer.application-id"
+						+ "=testKafkaStreamsBinderWithStandardConfigurationCanStart-foobar",
+				"--spring.cloud.stream.kafka.streams.binder.jaas.loginModule=org.apache.kafka.common.security.plain.PlainLoginModule",
+				"--spring.cloud.stream.kafka.streams.binder.jaas.options.username=foo",
+				"--spring.cloud.stream.kafka.streams.binder.jaas.options.password=bar",
+				"--spring.cloud.stream.kafka.streams.binder.brokers="
+						+ embeddedKafka.getEmbeddedKafka().getBrokersAsString());
+		javax.security.auth.login.Configuration configuration = javax.security.auth.login.Configuration
+				.getConfiguration();
+		final AppConfigurationEntry[] kafkaConfiguration = configuration
+				.getAppConfigurationEntry("KafkaClient");
+		assertThat(kafkaConfiguration).hasSize(1);
+		assertThat(kafkaConfiguration[0].getOptions().get("username")).isEqualTo("foo");
+		assertThat(kafkaConfiguration[0].getOptions().get("password")).isEqualTo("bar");
+		assertThat(kafkaConfiguration[0].getControlFlag())
+				.isEqualTo(AppConfigurationEntry.LoginModuleControlFlag.REQUIRED);
+		applicationContext.close();
+	}
+
 	@SpringBootApplication
 	@EnableBinding({SimpleKStreamBinding.class, SimpleKTableBinding.class, SimpleGlobalKTableBinding.class})
 	static class SimpleKafkaStreamsApplication {