[FLINK-6117] [security] Make setting of 'zookeeper.sasl.disable' work…

… correctly This closes apache#3600
xiaobabyLu · Apr 21, 2017 · eef85e0 · eef85e0
1 parent daf4038
commit eef85e0
Show file tree

Hide file tree

Showing 4 changed files with 22 additions and 0 deletions.
diff --git a/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java b/flink-core/src/main/java/org/apache/flink/configuration/SecurityOptions.java
@@ -55,6 +55,10 @@ public class SecurityOptions {
 	//  ZooKeeper Security Options
 	// ------------------------------------------------------------------------
 
+	public static final ConfigOption<Boolean> ZOOKEEPER_SASL_DISABLE =
+		key("zookeeper.sasl.disable")
+			.defaultValue(false);
+
 	public static final ConfigOption<String> ZOOKEEPER_SASL_SERVICE_NAME =
 		key("zookeeper.sasl.service-name")
 			.defaultValue("zookeeper");

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/security/SecurityUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/security/SecurityUtils.java
@@ -125,6 +125,8 @@ public static class SecurityConfiguration {
 
 		private final org.apache.hadoop.conf.Configuration hadoopConf;
 
+		private final boolean isZkSaslDisable;
+
 		private final boolean useTicketCache;
 
 		private final String keytab;
@@ -164,6 +166,7 @@ public SecurityConfiguration(Configuration flinkConf,
 				org.apache.hadoop.conf.Configuration hadoopConf,
 				List<? extends Class<? extends SecurityModule>> securityModules) {
 			this.hadoopConf = checkNotNull(hadoopConf);
+			this.isZkSaslDisable = flinkConf.getBoolean(SecurityOptions.ZOOKEEPER_SASL_DISABLE);
 			this.keytab = flinkConf.getString(SecurityOptions.KERBEROS_LOGIN_KEYTAB);
 			this.principal = flinkConf.getString(SecurityOptions.KERBEROS_LOGIN_PRINCIPAL);
 			this.useTicketCache = flinkConf.getBoolean(SecurityOptions.KERBEROS_LOGIN_USETICKETCACHE);
@@ -175,6 +178,10 @@ public SecurityConfiguration(Configuration flinkConf,
 			validate();
 		}
 
+		public boolean isZkSaslDisable() {
+			return isZkSaslDisable;
+		}
+
 		public String getKeytab() {
 			return keytab;
 		}

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/security/modules/ZooKeeperModule.java b/flink-runtime/src/main/java/org/apache/flink/runtime/security/modules/ZooKeeperModule.java
@@ -41,13 +41,18 @@ public class ZooKeeperModule implements SecurityModule {
 	 */
 	private static final String ZK_LOGIN_CONTEXT_NAME = "zookeeper.sasl.clientconfig";
 
+	private String priorSaslEnable;
+
 	private String priorServiceName;
 
 	private String priorLoginContextName;
 
 	@Override
 	public void install(SecurityUtils.SecurityConfiguration configuration) throws SecurityInstallException {
 
+		priorSaslEnable = System.getProperty(ZK_ENABLE_CLIENT_SASL, null);
+		System.setProperty(ZK_ENABLE_CLIENT_SASL, String.valueOf(!configuration.isZkSaslDisable()));
+
 		priorServiceName = System.getProperty(ZK_SASL_CLIENT_USERNAME, null);
 		if (!"zookeeper".equals(configuration.getZooKeeperServiceName())) {
 			System.setProperty(ZK_SASL_CLIENT_USERNAME, configuration.getZooKeeperServiceName());
@@ -61,6 +66,11 @@ public void install(SecurityUtils.SecurityConfiguration configuration) throws Se
 
 	@Override
 	public void uninstall() throws SecurityInstallException {
+		if(priorSaslEnable != null) {
+			System.setProperty(ZK_ENABLE_CLIENT_SASL, priorSaslEnable);
+		} else {
+			System.clearProperty(ZK_ENABLE_CLIENT_SASL);
+		}
 		if(priorServiceName != null) {
 			System.setProperty(ZK_SASL_CLIENT_USERNAME, priorServiceName);
 		} else {

diff --git a/...rent/flink-test-utils/src/main/java/org/apache/flink/test/util/SecureTestEnvironment.java b/...rent/flink-test-utils/src/main/java/org/apache/flink/test/util/SecureTestEnvironment.java
@@ -138,6 +138,7 @@ public static void prepare(TemporaryFolder tempFolder) {
 			//ctx.setHadoopConfiguration() for the UGI implementation to work properly.
 			//See Yarn test case module for reference
 			Configuration flinkConfig = GlobalConfiguration.loadConfiguration();
+			flinkConfig.setBoolean(SecurityOptions.ZOOKEEPER_SASL_DISABLE, false);
 			flinkConfig.setString(SecurityOptions.KERBEROS_LOGIN_KEYTAB, testKeytab);
 			flinkConfig.setBoolean(SecurityOptions.KERBEROS_LOGIN_USETICKETCACHE, false);
 			flinkConfig.setString(SecurityOptions.KERBEROS_LOGIN_PRINCIPAL, testPrincipal);