[kafka] branch 3.0 updated: MINOR: make sure alterAclsPurgatory is closed when controller server … (#10868)

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

chia7712 pushed a commit to branch 3.0
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/3.0 by this push:
     new 919b255  MINOR: make sure alterAclsPurgatory is closed when controller server … (#10868)
919b255 is described below

commit 919b2550403282a43482c38d168ecd95bd6f9fad
Author: Chia-Ping Tsai <ch...@gmail.com>
AuthorDate: Sat Jul 17 14:52:37 2021 +0800

    MINOR: make sure alterAclsPurgatory is closed when controller server … (#10868)
    
    Reviewers:  Ismael Juma <is...@juma.me.uk>
---
 core/src/main/scala/kafka/server/AclApis.scala                 | 6 +++---
 core/src/main/scala/kafka/server/ControllerApis.scala          | 4 ++++
 core/src/main/scala/kafka/server/ControllerServer.scala        | 2 ++
 core/src/main/scala/kafka/server/DelayedFuture.scala           | 2 ++
 core/src/test/scala/unit/kafka/server/ControllerApisTest.scala | 7 +++++++
 5 files changed, 18 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/kafka/server/AclApis.scala b/core/src/main/scala/kafka/server/AclApis.scala
index a79ce84..97b685b 100644
--- a/core/src/main/scala/kafka/server/AclApis.scala
+++ b/core/src/main/scala/kafka/server/AclApis.scala
@@ -49,9 +49,9 @@ class AclApis(authHelper: AuthHelper,
   private val alterAclsPurgatory =
       new DelayedFuturePurgatory(purgatoryName = "AlterAcls", brokerId = config.nodeId)
 
-  def close(): Unit = {
-    alterAclsPurgatory.shutdown()
-  }
+  def isClosed: Boolean = alterAclsPurgatory.isShutdown
+
+  def close(): Unit = alterAclsPurgatory.shutdown()
 
   def handleDescribeAcls(request: RequestChannel.Request): Unit = {
     authHelper.authorizeClusterOperation(request, DESCRIBE)
diff --git a/core/src/main/scala/kafka/server/ControllerApis.scala b/core/src/main/scala/kafka/server/ControllerApis.scala
index 0eddbb4..35028b4 100644
--- a/core/src/main/scala/kafka/server/ControllerApis.scala
+++ b/core/src/main/scala/kafka/server/ControllerApis.scala
@@ -77,6 +77,10 @@ class ControllerApis(val requestChannel: RequestChannel,
   val requestHelper = new RequestHandlerHelper(requestChannel, quotas, time)
   private val aclApis = new AclApis(authHelper, authorizer, requestHelper, "controller", config)
 
+  def isClosed: Boolean = aclApis.isClosed
+
+  def close(): Unit = aclApis.close()
+
   override def handle(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
     try {
       request.header.apiKey match {
diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala b/core/src/main/scala/kafka/server/ControllerServer.scala
index 1518661..2ac7861 100644
--- a/core/src/main/scala/kafka/server/ControllerServer.scala
+++ b/core/src/main/scala/kafka/server/ControllerServer.scala
@@ -199,6 +199,8 @@ class ControllerServer(
         CoreUtils.swallow(socketServer.shutdown(), this)
       if (controllerApisHandlerPool != null)
         CoreUtils.swallow(controllerApisHandlerPool.shutdown(), this)
+      if (controllerApis != null)
+        CoreUtils.swallow(controllerApis.close(), this)
       if (quotaManagers != null)
         CoreUtils.swallow(quotaManagers.shutdown(), this)
       if (controller != null)
diff --git a/core/src/main/scala/kafka/server/DelayedFuture.scala b/core/src/main/scala/kafka/server/DelayedFuture.scala
index cf522ab..018fae0 100644
--- a/core/src/main/scala/kafka/server/DelayedFuture.scala
+++ b/core/src/main/scala/kafka/server/DelayedFuture.scala
@@ -97,4 +97,6 @@ class DelayedFuturePurgatory(purgatoryName: String, brokerId: Int) {
     executor.awaitTermination(60, TimeUnit.SECONDS)
     purgatory.shutdown()
   }
+
+  def isShutdown: Boolean = executor.isShutdown
 }
diff --git a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
index 58a9479..247ddc4 100644
--- a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala
@@ -317,6 +317,13 @@ class ControllerApisTest {
   }
 
   @Test
+  def testClose(): Unit = {
+    val apis = createControllerApis(Some(createDenyAllAuthorizer()), mock(classOf[Controller]))
+    apis.close()
+    assertTrue(apis.isClosed)
+  }
+
+  @Test
   def testUnauthorizedBrokerRegistration(): Unit = {
     val brokerRegistrationRequest = new BrokerRegistrationRequest.Builder(
       new BrokerRegistrationRequestData()