ZOOKEEPER-5059: Kerberos SASL leaks Login refresh threads across QuorumPeer.shutdown()

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/QuorumPeer.java

@@ -1712,6 +1712,12 @@ public void shutdown() {
                 LOG.warn("Error closing logs ", ie);
             }
         }
+        if (authServer != null) {
+            authServer.shutdown();
+        }
+        if (authLearner != null) {
+            authLearner.shutdown();
+        }
     }
 
     /**

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/NullQuorumAuthLearner.java

@@ -31,4 +31,9 @@ public void authenticate(Socket sock, String hostname) {
         return; // simply return don't require auth
     }
 
+    @Override
+    public void shutdown() {
+        // no-op
+    }
+
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/NullQuorumAuthServer.java

@@ -32,4 +32,9 @@ public void authenticate(final Socket sock, final DataInputStream din) {
         // simply return don't require auth
     }
 
+    @Override
+    public void shutdown() {
+        // no-op
+    }
+
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/QuorumAuthLearner.java

@@ -38,4 +38,8 @@ public interface QuorumAuthLearner {
      */
     void authenticate(Socket sock, String hostname) throws IOException;
 
+    /**
+     * Shutdown the learner and release resources.
+     */
+    void shutdown();
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/QuorumAuthServer.java

@@ -38,4 +38,8 @@ public interface QuorumAuthServer {
      */
     void authenticate(Socket sock, DataInputStream din) throws IOException;
 
+    /**
+     * Shut down the server and release resources.
+     */
+    void shutdown();
 }

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/SaslQuorumAuthLearner.java

@@ -144,6 +144,13 @@ public void authenticate(Socket sock, String hostName) throws IOException {
         }
     }
 
+    @Override
+    public void shutdown() {
+        if (learnerLogin != null) {
+            learnerLogin.shutdown();
+        }
+    }
+
     private void checkAuthStatus(Socket sock, QuorumAuth.Status qpStatus) throws SaslException {
         if (qpStatus == QuorumAuth.Status.SUCCESS) {
             LOG.info(

zookeeper-server/src/main/java/org/apache/zookeeper/server/quorum/auth/SaslQuorumAuthServer.java

@@ -156,6 +156,13 @@ public void authenticate(Socket sock, DataInputStream din) throws SaslException
         }
     }
 
+    @Override
+    public void shutdown() {
+        if (serverLogin != null) {
+            serverLogin.shutdown();
+        }
+    }
+
     private byte[] receive(DataInputStream din) throws IOException {
         QuorumAuthPacket authPacket = new QuorumAuthPacket();
         BinaryInputArchive bia = BinaryInputArchive.getArchive(din);