[FLINK-6608] [security, config] Relax Kerberos login contexts parsing

This closes #3928.

flink-runtime/src/main/java/org/apache/flink/runtime/security/SecurityUtils.java

@@ -230,10 +230,14 @@ public class SecurityUtils {
 		}
 
 		private static List<String> parseList(String value) {
-			if(value == null) {
+			if(value == null || value.isEmpty()) {
 				return Collections.emptyList();
 			}
-			return Arrays.asList(value.split(","));
+
+			return Arrays.asList(value
+				.trim()
+				.replaceAll("(\\s*,+\\s*)+", ",")
+				.split(","));
 		}
 	}
 

flink-runtime/src/test/java/org/apache/flink/runtime/security/SecurityUtilsTest.java

@@ -18,13 +18,19 @@
 package org.apache.flink.runtime.security;
 
 import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.SecurityOptions;
 import org.apache.flink.runtime.security.modules.SecurityModule;
 import org.junit.AfterClass;
 import org.junit.Test;
 
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
 /**
  * Tests for the {@link SecurityUtils}.
@@ -78,4 +84,58 @@ public class SecurityUtilsTest {
 		SecurityUtils.uninstall();
 		assertEquals(NoOpSecurityContext.class, SecurityUtils.getInstalledContext().getClass());
 	}
+
+	@Test
+	public void testKerberosLoginContextParsing() {
+
+		List<String> expectedLoginContexts = Arrays.asList("Foo bar", "Client");
+
+		Configuration testFlinkConf;
+		SecurityUtils.SecurityConfiguration testSecurityConf;
+
+		// ------- no whitespaces
+
+		testFlinkConf = new Configuration();
+		testFlinkConf.setString(SecurityOptions.KERBEROS_LOGIN_CONTEXTS, "Foo bar,Client");
+		testSecurityConf = new SecurityUtils.SecurityConfiguration(
+			testFlinkConf, new org.apache.hadoop.conf.Configuration(),
+			Collections.singletonList(TestSecurityModule.class));
+		assertEquals(expectedLoginContexts, testSecurityConf.getLoginContextNames());
+
+		// ------- with whitespaces surrounding comma
+
+		testFlinkConf = new Configuration();
+		testFlinkConf.setString(SecurityOptions.KERBEROS_LOGIN_CONTEXTS, "Foo bar , Client");
+		testSecurityConf = new SecurityUtils.SecurityConfiguration(
+			testFlinkConf, new org.apache.hadoop.conf.Configuration(),
+			Collections.singletonList(TestSecurityModule.class));
+		assertEquals(expectedLoginContexts, testSecurityConf.getLoginContextNames());
+
+		// ------- leading / trailing whitespaces at start and end of list
+
+		testFlinkConf = new Configuration();
+		testFlinkConf.setString(SecurityOptions.KERBEROS_LOGIN_CONTEXTS, " Foo bar , Client ");
+		testSecurityConf = new SecurityUtils.SecurityConfiguration(
+			testFlinkConf, new org.apache.hadoop.conf.Configuration(),
+			Collections.singletonList(TestSecurityModule.class));
+		assertEquals(expectedLoginContexts, testSecurityConf.getLoginContextNames());
+
+		// ------- empty entries
+
+		testFlinkConf = new Configuration();
+		testFlinkConf.setString(SecurityOptions.KERBEROS_LOGIN_CONTEXTS, "Foo bar,,Client");
+		testSecurityConf = new SecurityUtils.SecurityConfiguration(
+			testFlinkConf, new org.apache.hadoop.conf.Configuration(),
+			Collections.singletonList(TestSecurityModule.class));
+		assertEquals(expectedLoginContexts, testSecurityConf.getLoginContextNames());
+
+		// ------- empty trailing String entries with whitespaces
+
+		testFlinkConf = new Configuration();
+		testFlinkConf.setString(SecurityOptions.KERBEROS_LOGIN_CONTEXTS, "Foo bar, ,, Client,");
+		testSecurityConf = new SecurityUtils.SecurityConfiguration(
+			testFlinkConf, new org.apache.hadoop.conf.Configuration(),
+			Collections.singletonList(TestSecurityModule.class));
+		assertEquals(expectedLoginContexts, testSecurityConf.getLoginContextNames());
+	}
 }