Revert "Remove KillProcessCompletedEvent"

mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/subscriber/ProcessListChangedSubscriber.java

@@ -28,6 +28,7 @@
 import org.apache.shardingsphere.mode.manager.ContextManager;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.RegistryCenter;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.KillProcessEvent;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.KillProcessCompletedEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.ReportLocalProcessesCompletedEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.ShowProcessListTriggerEvent;
 
@@ -99,6 +100,15 @@ public synchronized void killProcess(final KillProcessEvent event) throws SQLExc
             }
         }
         registryCenter.getRepository().delete(ComputeNode.getProcessKillInstanceIdNodePath(event.getInstanceId(), event.getProcessId()));
+    }
+
+    /**
+     * Complete to kill process.
+     *
+     * @param event kill process completed event
+     */
+    @Subscribe
+    public synchronized void completeToKillProcess(final KillProcessCompletedEvent event) {
         ProcessOperationLockRegistry.getInstance().notify(event.getProcessId());
     }
 }

mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/event/KillProcessCompletedEvent.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.GovernanceEvent;
+
+/**
+ * Kill process completed event.
+ */
+@RequiredArgsConstructor
+@Getter
+public final class KillProcessCompletedEvent implements GovernanceEvent {
+
+    private final String processId;
+}

mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/status/compute/watcher/ComputeNodeStateChangedWatcher.java

@@ -29,6 +29,7 @@
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.InstanceOfflineEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.InstanceOnlineEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.KillProcessEvent;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.KillProcessCompletedEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.LabelsEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.ReportLocalProcessesCompletedEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.ShowProcessListTriggerEvent;
@@ -133,6 +134,9 @@ private Optional<GovernanceEvent> createKillProcessIdEvent(final DataChangedEven
         if (Type.ADDED == event.getType()) {
             return Optional.of(new KillProcessEvent(matcher.group(1), matcher.group(2)));
         }
+        if (Type.DELETED == event.getType()) {
+            return Optional.of(new KillProcessCompletedEvent(matcher.group(2)));
+        }
         return Optional.empty();
     }
 

mode/type/cluster/core/src/test/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/subscriber/ProcessListChangedSubscriberTest.java

@@ -36,6 +36,7 @@
 import org.apache.shardingsphere.mode.manager.cluster.ClusterContextManagerBuilder;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.RegistryCenter;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.KillProcessEvent;
+import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.KillProcessCompletedEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.ReportLocalProcessesCompletedEvent;
 import org.apache.shardingsphere.mode.manager.cluster.coordinator.registry.status.compute.event.ShowProcessListTriggerEvent;
 import org.apache.shardingsphere.mode.metadata.MetaDataContexts;
@@ -141,6 +142,23 @@ void assertKillProcess() throws SQLException, ReflectiveOperationException {
         verify(repository).delete("/nodes/compute_nodes/process_kill/" + instanceId + ":foo_id");
     }
 
+    @Test
+    void assertCompleteToKillProcess() {
+        String processId = "foo_id";
+        long startMillis = System.currentTimeMillis();
+        Executors.newFixedThreadPool(1).submit(() -> {
+            try {
+                Thread.sleep(50L);
+            } catch (final InterruptedException ignored) {
+            }
+            subscriber.completeToKillProcess(new KillProcessCompletedEvent(processId));
+        });
+        waitUntilReleaseReady(processId);
+        long currentMillis = System.currentTimeMillis();
+        assertThat(currentMillis, greaterThanOrEqualTo(startMillis + 50L));
+        assertThat(currentMillis, lessThanOrEqualTo(startMillis + 5000L));
+    }
+
     private static void waitUntilReleaseReady(final String lockId) {
         ProcessOperationLockRegistry.getInstance().waitUntilReleaseReady(lockId, new ProcessOperationLockReleaseStrategy() {