fix(issues1248): fix partition reassignment deadlock (#1250)

Signed-off-by: Robin Han <[email protected]>
AutoMQ · May 14, 2024 · 35ec4ee · 35ec4ee
1 parent 671d2ee
commit 35ec4ee
Showing 1 changed file with 6 additions and 4 deletions.
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -349,7 +349,8 @@ class ReplicaManager(val config: KafkaConfig,
   /**
    * Partition operation executor, used to execute partition operations in parallel.
    */
-  private val partitionOpExecutor = ThreadUtils.newCachedThread(256, "partition_op_%d", true)
+  private val partitionOpenOpExecutor = ThreadUtils.newCachedThread(128, "partition_open_op_%d", true)
+  private val partitionCloseOpExecutor = ThreadUtils.newCachedThread(128, "partition_close_op_%d", true)
   /**
    * Partition operation map, used to make sure that only one operation is executed for a partition at the same time.
    * It should be modified with [[replicaStateChangeLock]] held.
@@ -2732,7 +2733,7 @@ class ReplicaManager(val config: KafkaConfig,
               partitionOpMap.put(tp, opCf)
               openingPartitions.putIfAbsent(tp, opCf)
               prevOp.whenComplete((_, _) => {
-                partitionOpExecutor.execute(() => {
+                partitionOpenOpExecutor.execute(() => {
                   try {
                     val leader = mutable.Map[TopicPartition, LocalReplicaChanges.PartitionInfo]()
                     leader += (tp -> info)
@@ -2792,7 +2793,7 @@ class ReplicaManager(val config: KafkaConfig,
     partitionOpMap.put(tp, opCf)
     closingPartitions.put(tp, opCf)
     prevOp.whenComplete((_, _) => {
-      partitionOpExecutor.execute(() => {
+      partitionCloseOpExecutor.execute(() => {
         try {
           val delete = mutable.Map[TopicPartition, Boolean]()
           delete += (tp -> true)
@@ -2974,7 +2975,8 @@ class ReplicaManager(val config: KafkaConfig,
     } else {
       info(s"all partitions are closed after ${System.currentTimeMillis() - start} ms")
     }
-    partitionOpExecutor.shutdown()
+    partitionOpenOpExecutor.shutdown()
+    partitionCloseOpExecutor.shutdown()
     CoreUtils.swallow(ElasticLogManager.shutdown(), this)
   }