diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java index 7eae7953d..e18e338fc 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/GlobalKTableBinderConfiguration.java +++ b/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 diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java index 9450bc392..26d094797 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamBinderConfiguration.java +++ b/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 diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java index 825e4a2ae..50112f58f 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KTableBinderConfiguration.java +++ b/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 diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsJaasConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsJaasConfiguration.java new file mode 100644 index 000000000..8ee3d5257 --- /dev/null +++ b/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; + } +} diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java index 5ed9c0b64..ea977a4e7 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/bootstrap/KafkaStreamsBinderBootstrapTest.java +++ b/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 {