From 30c868e51b6d563d8488948476947c50997da4ff Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Fri, 16 Sep 2016 11:26:07 -0400 Subject: [PATCH] Configure JAAS settings for Kafka via Spring Boot Fixes #38 Adds support for programmatic configurtion of JAAS properties as an alternative to using a JAAS file. Add a placeholder JAAS configuration file which is ignored by the configuration but lets Kafka 0.9 clients connect --- .../src/main/asciidoc/overview.adoc | 75 +++++++++-- .../KafkaBinderJaasInitializerListener.java | 118 ++++++++++++++++++ .../kafka/KafkaMessageChannelBinder.java | 2 +- .../config/JaasLoginModuleConfiguration.java | 82 ++++++++++++ .../config/KafkaBinderConfiguration.java | 16 +++ .../KafkaBinderConfigurationProperties.java | 11 ++ ...afkaBinderJaasInitializerListenerTest.java | 60 +++++++++ .../resources/jaas-sample-kafka-only.conf | 7 ++ .../test/resources/jaas-sample-with-zk.conf | 14 +++ 9 files changed, 377 insertions(+), 8 deletions(-) create mode 100644 spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListener.java create mode 100644 spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/JaasLoginModuleConfiguration.java create mode 100644 spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java create mode 100644 spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-kafka-only.conf create mode 100644 spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-with-zk.conf diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc index 0356783ee..d4fdaddb2 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/overview.adoc @@ -215,18 +215,78 @@ spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_SSL All the other security properties can be set in a similar manner. When using Kerberos, follow the instructions in the http://kafka.apache.org/090/documentation.html#security_sasl_clientconfig[reference documentation] for creating and referencing the JAAS configuration. -At the time of this release, the JAAS, and (optionally) krb5 file locations must be set for Spring Cloud Stream applications by using system properties. -Here is an example of launching a Spring Cloud Stream application with SASL and Kerberos. + +Spring Cloud Stream supports passing JAAS configuration information to the application using a JAAS configuration file and using Spring Boot properties. + +===== Using JAAS configuration files + +The JAAS, and (optionally) krb5 file locations can be set for Spring Cloud Stream applications by using system properties. +Here is an example of launching a Spring Cloud Stream application with SASL and Kerberos using a JAAS configuration file: [source] ---- - java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \\ - --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \\ - --spring.cloud.stream.kafka.binder.zkNodes=secure.zookeeper:2181 \\ - --spring.cloud.stream.bindings.input.destination=stream.ticktock \\ - --spring.cloud.stream.kafka.binder.clientConfiguration.security.protocol=SASL_PLAINTEXT + java -Djava.security.auth.login.config=/path.to/kafka_client_jaas.conf -jar log.jar \ + --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \ + --spring.cloud.stream.kafka.binder.zkNodes=secure.zookeeper:2181 \ + --spring.cloud.stream.bindings.input.destination=stream.ticktock \ + --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT ---- +===== Using Spring Boot properties + +As an alternative to having a JAAS configuration file, Spring Cloud Stream provides a mechanism for setting up the JAAS configuration for Spring Cloud Stream applications using Spring Boot properties. + +The following properties can be used for configuring the login context of the Kafka client. + +spring.cloud.stream.kafka.binder.jaas.loginModule:: + The login module name. Not necessary to be set in normal cases. ++ +Default: `com.sun.security.auth.module.Krb5LoginModule`. +spring.cloud.stream.kafka.binder.jaas.controlFlag:: + The control flag of the login module. ++ +Default: `required`. +spring.cloud.stream.kafka.binder.jaas.options:: + Map with a key/value pair containing the login module options. ++ +Default: Empty map. + +Here is an example of launching a Spring Cloud Stream application with SASL and Kerberos using Spring Boot configuration properties: + +[source] +---- + java --spring.cloud.stream.kafka.binder.brokers=secure.server:9092 \ + --spring.cloud.stream.kafka.binder.zkNodes=secure.zookeeper:2181 \ + --spring.cloud.stream.bindings.input.destination=stream.ticktock \ + --spring.cloud.stream.kafka.binder.autoCreateTopics=false \ + --spring.cloud.stream.kafka.binder.configuration.security.protocol=SASL_PLAINTEXT \ + --spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true \ + --spring.cloud.stream.kafka.binder.jaas.options.storeKey=true \ + --spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab \ + --spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM +---- + +This represents the equivalent of the following JAAS file: + +[source] +---- +KafkaClient { + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + keyTab="/etc/security/keytabs/kafka_client.keytab" + principal="kafka-client-1@EXAMPLE.COM"; +}; +---- + +If the topics required already exist on the broker, or will be created by an administrator, autocreation can be turned off and only client JAAS properties need to be sent. As an alternative to setting `spring.cloud.stream.kafka.binder.autoCreateTopics` you can simply remove the broker dependency from the application. See <> for details. + +[NOTE] +==== +Do not mix JAAS configuration files and Spring Boot properties in the same application. +If the `-Djava.security.auth.login.config` system property is already present, Spring Cloud Stream will ignore the Spring Boot properties. + +==== [NOTE] ==== @@ -272,6 +332,7 @@ The versions above are provided only for the sake of the example. For best results, we recommend using the most recent 0.10-compatible versions of the projects. ==== +[[exclude-admin-utils]] ==== Excluding Kafka broker jar from the classpath of the binder based application The Apache Kafka Binder uses the administrative utilities which are part of the Apache Kafka server library to create and reconfigure topics. diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListener.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListener.java new file mode 100644 index 000000000..c5e649825 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListener.java @@ -0,0 +1,118 @@ +/* + * Copyright 2016 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 + * + * http://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; + +import java.io.File; +import java.io.IOException; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import javax.security.auth.login.AppConfigurationEntry; +import javax.security.auth.login.Configuration; + +import org.apache.kafka.common.security.JaasUtils; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.cloud.stream.binder.kafka.config.KafkaBinderConfigurationProperties; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.context.ApplicationListener; +import org.springframework.context.event.ContextRefreshedEvent; +import org.springframework.util.Assert; + +/** + * @author Marius Bogoevici + */ +public class KafkaBinderJaasInitializerListener implements ApplicationListener, + ApplicationContextAware, DisposableBean { + + public static final String DEFAULT_ZK_LOGIN_CONTEXT_NAME = "Client"; + + private ApplicationContext applicationContext; + + private final boolean ignoreJavaLoginConfigParamSystemProperty; + + private final File placeholderJaasConfiguration; + + public KafkaBinderJaasInitializerListener() throws IOException { + // we ignore the system property if it wasn't originally set at launch + this.ignoreJavaLoginConfigParamSystemProperty = + (System.getProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM) == null); + this.placeholderJaasConfiguration = File.createTempFile("kafka-client-jaas-config-placeholder", "conf"); + this.placeholderJaasConfiguration.deleteOnExit(); + } + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + this.applicationContext = applicationContext; + } + + @Override + public void destroy() throws Exception { + if (this.ignoreJavaLoginConfigParamSystemProperty) { + System.clearProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM); + } + } + + @Override + public void onApplicationEvent(ContextRefreshedEvent event) { + if (event.getSource() == this.applicationContext) { + KafkaBinderConfigurationProperties binderConfigurationProperties = + applicationContext.getBean(KafkaBinderConfigurationProperties.class); + // only use programmatic support if a file is not set via system property + if (ignoreJavaLoginConfigParamSystemProperty + && binderConfigurationProperties.getJaas() != null) { + Map configurationEntries = new HashMap<>(); + AppConfigurationEntry kafkaClientConfigurationEntry = new AppConfigurationEntry + (binderConfigurationProperties.getJaas().getLoginModule(), + binderConfigurationProperties.getJaas().getControlFlagValue(), + binderConfigurationProperties.getJaas().getOptions() != null ? + binderConfigurationProperties.getJaas().getOptions() : + Collections.emptyMap()); + configurationEntries.put(JaasUtils.LOGIN_CONTEXT_CLIENT, + new AppConfigurationEntry[]{ kafkaClientConfigurationEntry }); + Configuration.setConfiguration(new InternalConfiguration(configurationEntries)); + // Workaround for a 0.9 client issue where even if the Configuration is set + // a system property check is performed. + // Since the Configuration already exists, this will be ignored. + if (this.placeholderJaasConfiguration != null) { + System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, this.placeholderJaasConfiguration.getAbsolutePath()); + } + } + } + } + + /** + * A {@link Configuration} set up programmatically by the Kafka binder + */ + public static class InternalConfiguration extends Configuration { + + private final Map configurationEntries; + + public InternalConfiguration(Map configurationEntries) { + Assert.notNull(configurationEntries, " cannot be null"); + Assert.notEmpty(configurationEntries, " cannot be empty"); + this.configurationEntries = configurationEntries; + } + + @Override + public AppConfigurationEntry[] getAppConfigurationEntry(String name) { + return configurationEntries.get(name); + } + } +} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 7fc2b864c..9894872e8 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -436,7 +436,7 @@ public class KafkaMessageChannelBinder extends "No topic will be created by the binder"); } else if (!this.configurationProperties.isAutoCreateTopics()) { - this.logger.warn("Auto creation of topics is disabled."); + this.logger.info("Auto creation of topics is disabled."); } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/JaasLoginModuleConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/JaasLoginModuleConfiguration.java new file mode 100644 index 000000000..fc2e97c21 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/JaasLoginModuleConfiguration.java @@ -0,0 +1,82 @@ +/* + * Copyright 2016 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 + * + * http://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.config; + +import java.util.HashMap; +import java.util.Map; +import javax.security.auth.login.AppConfigurationEntry; + +import org.springframework.util.Assert; + +/** + * Contains properties for setting up an {@link AppConfigurationEntry} that can be used + * for the Kafka or Zookeeper client. + * + * @author Marius Bogoevici + */ +public class JaasLoginModuleConfiguration { + + private String loginModule = "com.sun.security.auth.module.Krb5LoginModule"; + + private AppConfigurationEntry.LoginModuleControlFlag controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUIRED; + + private Map options = new HashMap<>(); + + public String getLoginModule() { + return loginModule; + } + + public void setLoginModule(String loginModule) { + Assert.notNull(loginModule, "cannot be null"); + this.loginModule = loginModule; + } + + public String getControlFlag() { + return controlFlag.toString(); + } + + public AppConfigurationEntry.LoginModuleControlFlag getControlFlagValue() { + return controlFlag; + } + + public void setControlFlag(String controlFlag) { + Assert.notNull(controlFlag, "cannot be null"); + if (AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL.equals(controlFlag)) { + this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.OPTIONAL; + } + else if (AppConfigurationEntry.LoginModuleControlFlag.REQUIRED.equals(controlFlag)) { + this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUIRED; + } + else if (AppConfigurationEntry.LoginModuleControlFlag.REQUISITE.equals(controlFlag)) { + this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.REQUISITE; + } + else if (AppConfigurationEntry.LoginModuleControlFlag.SUFFICIENT.equals(controlFlag)) { + this.controlFlag = AppConfigurationEntry.LoginModuleControlFlag.SUFFICIENT; + } + else { + throw new IllegalArgumentException(controlFlag + " is not a supported control flag"); + } + } + + public Map getOptions() { + return options; + } + + public void setOptions(Map options) { + this.options = options; + } +} diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java index 15e5d750b..f80a46dde 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfiguration.java @@ -16,6 +16,8 @@ package org.springframework.cloud.stream.binder.kafka.config; +import java.io.IOException; + import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.apache.kafka.common.utils.AppInfoParser; @@ -27,6 +29,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.Binder; import org.springframework.cloud.stream.binder.kafka.KafkaBinderHealthIndicator; +import org.springframework.cloud.stream.binder.kafka.KafkaBinderJaasInitializerListener; import org.springframework.cloud.stream.binder.kafka.KafkaExtendedBindingProperties; import org.springframework.cloud.stream.binder.kafka.KafkaMessageChannelBinder; import org.springframework.cloud.stream.binder.kafka.admin.AdminUtilsOperation; @@ -34,6 +37,7 @@ import org.springframework.cloud.stream.binder.kafka.admin.Kafka09AdminUtilsOper import org.springframework.cloud.stream.binder.kafka.admin.Kafka10AdminUtilsOperation; import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration; import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationListener; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Condition; import org.springframework.context.annotation.ConditionContext; @@ -116,6 +120,11 @@ public class KafkaBinderConfiguration { return new Kafka10AdminUtilsOperation(); } + @Bean + public ApplicationListener jaasInitializer() throws IOException { + return new KafkaBinderJaasInitializerListener(); + } + static class Kafka10Present implements Condition { @Override @@ -131,4 +140,11 @@ public class KafkaBinderConfiguration { return AppInfoParser.getVersion().startsWith("0.9"); } } + + public static class JaasConfigurationProperties { + + private JaasLoginModuleConfiguration kafka; + + private JaasLoginModuleConfiguration zookeeper; + } } diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java index 6c9e1467f..158073a6d 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/config/KafkaBinderConfigurationProperties.java @@ -77,6 +77,8 @@ public class KafkaBinderConfigurationProperties { private int queueSize = 8192; + private JaasLoginModuleConfiguration jaas; + public String getZkConnectionString() { return toConnectionString(this.zkNodes, this.defaultZkPort); } @@ -254,4 +256,13 @@ public class KafkaBinderConfigurationProperties { public void setConfiguration(Map configuration) { this.configuration = configuration; } + + public JaasLoginModuleConfiguration getJaas() { + return jaas; + } + + public void setJaas(JaasLoginModuleConfiguration jaas) { + this.jaas = jaas; + } + } diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java new file mode 100644 index 000000000..e833b6616 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderJaasInitializerListenerTest.java @@ -0,0 +1,60 @@ +/* + * Copyright 2016 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 + * + * http://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; + +import javax.security.auth.login.AppConfigurationEntry; + +import com.sun.security.auth.login.ConfigFile; +import org.apache.kafka.common.security.JaasUtils; +import org.junit.Test; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.core.io.ClassPathResource; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Marius Bogoevici + */ +public class KafkaBinderJaasInitializerListenerTest { + + @Test + public void testConfigurationParsedCorrectlyWithKafkaClient() throws Exception { + ConfigFile configFile = new ConfigFile(new ClassPathResource("jaas-sample-kafka-only.conf").getURI()); + final AppConfigurationEntry[] kafkaConfigurationArray = configFile.getAppConfigurationEntry(JaasUtils.LOGIN_CONTEXT_CLIENT); + + final ConfigurableApplicationContext context = + SpringApplication.run(SimpleApplication.class, + "--spring.cloud.stream.kafka.binder.jaas.options.useKeyTab=true", + "--spring.cloud.stream.kafka.binder.jaas.options.storeKey=true", + "--spring.cloud.stream.kafka.binder.jaas.options.keyTab=/etc/security/keytabs/kafka_client.keytab", + "--spring.cloud.stream.kafka.binder.jaas.options.principal=kafka-client-1@EXAMPLE.COM"); + javax.security.auth.login.Configuration configuration = javax.security.auth.login.Configuration.getConfiguration(); + + final AppConfigurationEntry[] kafkaConfiguration = configuration.getAppConfigurationEntry(JaasUtils.LOGIN_CONTEXT_CLIENT); + assertThat(kafkaConfiguration).hasSize(1); + assertThat(kafkaConfiguration[0].getOptions()).isEqualTo(kafkaConfigurationArray[0].getOptions()); + context.close(); + } + + @SpringBootApplication + public static class SimpleApplication { + + } +} diff --git a/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-kafka-only.conf b/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-kafka-only.conf new file mode 100644 index 000000000..9b6891fb8 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-kafka-only.conf @@ -0,0 +1,7 @@ +KafkaClient { + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + keyTab="/etc/security/keytabs/kafka_client.keytab" + principal="kafka-client-1@EXAMPLE.COM"; +}; \ No newline at end of file diff --git a/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-with-zk.conf b/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-with-zk.conf new file mode 100644 index 000000000..fe511bd62 --- /dev/null +++ b/spring-cloud-stream-binder-kafka/src/test/resources/jaas-sample-with-zk.conf @@ -0,0 +1,14 @@ +KafkaClient { + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + keyTab="/etc/security/keytabs/kafka_client.keytab" + principal="kafka-client-1@EXAMPLE.COM"; +}; +Client { + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + keyTab="/etc/security/keytabs/zk_client.keytab" + principal="zk-client-1@EXAMPLE.COM"; +}; \ No newline at end of file