Fix internal procedure deduplication (#18036)

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/ProcedureExecutor.java

@@ -298,7 +298,6 @@ public void addInternalProcedure(InternalProcedure interalProcedure) {
     if (interalProcedure == null) {
       return;
     }
-    interalProcedure.setState(ProcedureState.WAITING_TIMEOUT);
     timeoutExecutor.add(interalProcedure);
   }
 

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/TimeoutExecutorThread.java

@@ -20,10 +20,13 @@
 package org.apache.iotdb.confignode.procedure;
 
 import org.apache.iotdb.confignode.i18n.ProcedureMessages;
+import org.apache.iotdb.confignode.procedure.state.ProcedureState;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.DelayQueue;
 import java.util.concurrent.Delayed;
 import java.util.concurrent.TimeUnit;
@@ -34,6 +37,7 @@ public class TimeoutExecutorThread<Env> extends StoppableThread {
   private static final int DELAY_QUEUE_TIMEOUT = 20;
   private final ProcedureExecutor<Env> executor;
   private final DelayQueue<ProcedureDelayContainer<Env>> queue = new DelayQueue<>();
+  private final Set<Procedure<Env>> registeredInternalProcedures = ConcurrentHashMap.newKeySet();
 
   public TimeoutExecutorThread(
       ProcedureExecutor<Env> envProcedureExecutor, ThreadGroup threadGroup, String name) {
@@ -44,12 +48,21 @@ public TimeoutExecutorThread(
 
   public void add(Procedure<Env> procedure) {
     ProcedureDelayContainer<Env> delayTask = new ProcedureDelayContainer<>(procedure);
+    if (procedure instanceof InternalProcedure) {
+      if (!registeredInternalProcedures.add(procedure)) {
+        return;
+      }
+      procedure.setState(ProcedureState.WAITING_TIMEOUT);
+    }
     queue.remove(delayTask);
     queue.add(delayTask);
   }
 
   public boolean remove(Procedure<Env> procedure) {
-    return queue.remove(new ProcedureDelayContainer<>(procedure)) || procedure.isFinished();
+    boolean unregistered =
+        procedure instanceof InternalProcedure && registeredInternalProcedures.remove(procedure);
+    boolean removed = queue.remove(new ProcedureDelayContainer<>(procedure));
+    return unregistered || removed || procedure.isFinished();
   }
 
   private ProcedureDelayContainer<Env> takeQuietly() {
@@ -70,12 +83,12 @@ public void run() {
       }
       Procedure<Env> procedure = delayTask.getProcedure();
       if (procedure instanceof InternalProcedure) {
-        if (procedure.isFinished()) {
+        if (!registeredInternalProcedures.contains(procedure) || procedure.isFinished()) {
           continue;
         }
         InternalProcedure internal = (InternalProcedure) procedure;
         internal.periodicExecute(executor.getEnvironment());
-        if (!procedure.isFinished()) {
+        if (!procedure.isFinished() && registeredInternalProcedures.contains(procedure)) {
           procedure.updateTimestamp();
           queue.add(delayTask);
         }

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/TestProcedureExecutor.java

@@ -127,6 +127,16 @@ public void testInternalProcedureCanBeDeduplicatedAndRemoved() throws Interrupte
     Assert.assertFalse(internalProcedure.awaitExecution(300, TimeUnit.MILLISECONDS));
     Assert.assertEquals(1, internalProcedure.getExecutionCount());
 
+    procExecutor.addInternalProcedure(internalProcedure);
+    Assert.assertFalse(internalProcedure.awaitExecution(300, TimeUnit.MILLISECONDS));
+    Assert.assertEquals(1, internalProcedure.getExecutionCount());
+
+    Assert.assertTrue(procExecutor.removeInternalProcedure(internalProcedure));
+
+    procExecutor.addInternalProcedure(internalProcedure);
+    Assert.assertTrue(internalProcedure.awaitExecution(30, TimeUnit.SECONDS));
+    Assert.assertEquals(2, internalProcedure.getExecutionCount());
+
     Assert.assertTrue(procExecutor.removeInternalProcedure(internalProcedure));
   }