KAFKA-5227; Remove unsafe assertion, make jaas config safer

Author: Rajini Sivaram <rajinisivaram@googlemail.com>

Reviewers: Ismael Juma <ismael@juma.me.uk>

Closes #3037 from rajinisivaram/KAFKA-5227

clients/src/test/java/org/apache/kafka/common/security/JaasContextTest.java

@@ -53,8 +53,8 @@ public class JaasContextTest {
     public void setUp() throws IOException {
         jaasConfigFile = File.createTempFile("jaas", ".conf");
         jaasConfigFile.deleteOnExit();
-        Configuration.setConfiguration(null);
         System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, jaasConfigFile.toString());
+        Configuration.setConfiguration(null);
     }
 
     @After

core/src/test/scala/integration/kafka/api/SaslSetup.scala

@@ -100,10 +100,10 @@ trait SaslSetup {
   }
 
   private def writeJaasConfigurationToFile(jaasSections: Seq[JaasSection]) {
-    // This will cause a reload of the Configuration singleton when `getConfiguration` is called
-    Configuration.setConfiguration(null)
     val file = JaasTestUtils.writeJaasContextsToFile(jaasSections)
     System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, file.getAbsolutePath)
+    // This will cause a reload of the Configuration singleton when `getConfiguration` is called
+    Configuration.setConfiguration(null)
   }
 
   def closeSasl() {

core/src/test/scala/unit/kafka/security/auth/ZkAuthorizationTest.scala

@@ -35,8 +35,8 @@ class ZkAuthorizationTest extends ZooKeeperTestHarness with Logging {
 
   @Before
   override def setUp() {
-    Configuration.setConfiguration(null)
     System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, jaasFile.getAbsolutePath)
+    Configuration.setConfiguration(null)
     System.setProperty(authProvider, "org.apache.zookeeper.server.auth.SASLAuthenticationProvider")
     super.setUp()
   }

core/src/test/scala/unit/kafka/zk/ZKEphemeralTest.scala

@@ -54,8 +54,8 @@ class ZKEphemeralTest(val secure: Boolean) extends ZooKeeperTestHarness {
   @Before
   override def setUp() {
     if (secure) {
-      Configuration.setConfiguration(null)
       System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, jaasFile.getAbsolutePath)
+      Configuration.setConfiguration(null)
       System.setProperty(authProvider, "org.apache.zookeeper.server.auth.SASLAuthenticationProvider")
       if (!JaasUtils.isZkSecurityEnabled)
         fail("Secure access not enabled")

core/src/test/scala/unit/kafka/zk/ZooKeeperTestHarness.scala

@@ -21,14 +21,11 @@ import javax.security.auth.login.Configuration
 
 import kafka.utils.{CoreUtils, Logging, ZkUtils}
 import org.junit.{After, Before}
-import org.junit.Assert.assertEquals
 import org.scalatest.junit.JUnitSuite
 import org.apache.kafka.common.security.JaasUtils
 import org.apache.kafka.test.IntegrationTest
 import org.junit.experimental.categories.Category
 
-import scala.collection.JavaConverters._
-
 @Category(Array(classOf[IntegrationTest]))
 abstract class ZooKeeperTestHarness extends JUnitSuite with Logging {
 
@@ -44,7 +41,6 @@ abstract class ZooKeeperTestHarness extends JUnitSuite with Logging {
   
   @Before
   def setUp() {
-    assertNoBrokerControllersRunning()
     zookeeper = new EmbeddedZookeeper()
     zkUtils = ZkUtils(zkConnect, zkSessionTimeout, zkConnectionTimeout, zkAclsEnabled.getOrElse(JaasUtils.isZkSecurityEnabled()))
   }
@@ -56,22 +52,5 @@ abstract class ZooKeeperTestHarness extends JUnitSuite with Logging {
     if (zookeeper != null)
       CoreUtils.swallow(zookeeper.shutdown())
     Configuration.setConfiguration(null)
-    assertNoBrokerControllersRunning()
-  }
-
-  // Tests using this class start ZooKeeper before starting any brokers and shutdown ZK after
-  // shutting down brokers. If tests leave broker controllers running, subsequent tests may fail in
-  // unexpected ways if ZK port is reused. This method ensures that there is no Controller event thread
-  // since the controller loads default JAAS configuration to make connections to brokers on this thread.
-  //
-  // Any tests that use this class and invoke ZooKeeperTestHarness#tearDown() will fail in the tearDown()
-  // if controller event thread is found. Tests with missing broker shutdown which don't use ZooKeeperTestHarness
-  // or its tearDown() will cause an assertion failure in the subsequent test that invokes ZooKeeperTestHarness#setUp(),
-  // making it easier to identify the test with missing shutdown from the test sequence.
-  private def assertNoBrokerControllersRunning() {
-    val threads = Thread.getAllStackTraces.keySet.asScala
-      .map(_.getName)
-      .filter(_.contains("controller-event-thread"))
-    assertEquals(Set(), threads)
   }
 }