(kafka) branch 4.0 updated: KAFKA-18790 Fix testCustomQuotaCallback (#18906)

[ch...@apache.org]

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

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


The following commit(s) were added to refs/heads/4.0 by this push:
     new 6abb4775b90 KAFKA-18790 Fix testCustomQuotaCallback (#18906)
6abb4775b90 is described below

commit 6abb4775b90465305bca2af344056cd828db7af8
Author: Ming-Yen Chung <mi...@gmail.com>
AuthorDate: Sat Feb 15 03:07:59 2025 +0800

    KAFKA-18790 Fix testCustomQuotaCallback (#18906)
    
    Frequently updating the trust store can cause unexpected termination of the AsyncConsumer background thread.
    
    1. To resolve this issue, reuse the same AdminClient instead of recreating it.
    2. Add error logging when fail to initialize resources for the consumer network thread.
    
    Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
 .../consumer/internals/ConsumerNetworkThread.java  |  2 ++
 .../kafka/api/CustomQuotaCallbackTest.scala        | 24 +++++++++-------------
 2 files changed, 12 insertions(+), 14 deletions(-)

diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
index 0e7b58acc21..a48289919b0 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java
@@ -109,6 +109,8 @@ public class ConsumerNetworkThread extends KafkaThread implements Closeable {
                     log.error("Unexpected error caught in consumer network thread", e);
                 }
             }
+        } catch (final Throwable e) {
+            log.error("Failed to initialize resources for consumer network thread", e);
         } finally {
             cleanup();
         }
diff --git a/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala b/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala
index 3d09c667371..006a5d085c5 100644
--- a/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala
+++ b/core/src/test/scala/integration/kafka/api/CustomQuotaCallbackTest.scala
@@ -40,9 +40,7 @@ import java.util.Properties
 import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
 import java.{lang, util}
-import scala.collection.mutable.ArrayBuffer
 import scala.jdk.CollectionConverters._
-import scala.util.Using
 
 class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {
 
@@ -57,8 +55,8 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {
   private val kafkaClientSaslMechanism = "SCRAM-SHA-256"
   override protected val serverSaslProperties = Some(kafkaServerSaslProperties(kafkaServerSaslMechanisms, kafkaClientSaslMechanism))
   override protected val clientSaslProperties = Some(kafkaClientSaslProperties(kafkaClientSaslMechanism))
-  private val adminClients = new ArrayBuffer[Admin]()
   private var producerWithoutQuota: KafkaProducer[Array[Byte], Array[Byte]] = _
+  private var admin: Admin = _
 
   val defaultRequestQuota = 1000
   val defaultProduceQuota = 2000 * 1000 * 1000
@@ -82,7 +80,7 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {
 
   @AfterEach
   override def tearDown(): Unit = {
-    adminClients.foreach(_.close())
+    if (admin != null) admin.close()
     GroupedUserQuotaCallback.tearDown()
     super.tearDown()
     closeSasl()
@@ -196,13 +194,11 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {
     topic: String, 
     listenerName: ListenerName = listenerName
   ): Unit = {
-    Using.resource(createAdminClient()) { admin =>
-      TestUtils.deleteTopicWithAdmin(
-        admin = admin,
-        topic = topic,
-        brokers = aliveBrokers,
-        controllers = controllerServers)
-    }
+    TestUtils.deleteTopicWithAdmin(
+      admin = createAdminClient(),
+      topic = topic,
+      brokers = aliveBrokers,
+      controllers = controllerServers)
   }
 
   private def createTopic(topic: String, numPartitions: Int, leader: Int): Unit = {
@@ -218,6 +214,7 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {
   }
 
   private def createAdminClient(): Admin = {
+    if (admin != null) return admin
     val config = new util.HashMap[String, Object]
     config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers())
     clientSecurityProps("admin-client").asInstanceOf[util.Map[Object, Object]].forEach { (key, value) =>
@@ -225,9 +222,8 @@ class CustomQuotaCallbackTest extends IntegrationTestHarness with SaslSetup {
     }
     config.put(SaslConfigs.SASL_JAAS_CONFIG,
       JaasModule.scramLoginModule(JaasTestUtils.KAFKA_SCRAM_ADMIN, JaasTestUtils.KAFKA_SCRAM_ADMIN_PASSWORD).toString)
-    val adminClient = Admin.create(config)
-    adminClients += adminClient
-    adminClient
+    admin = Admin.create(config)
+    admin
   }
 
   private def produceWithoutThrottle(topic: String, numRecords: Int): Unit = {