You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by zh...@apache.org on 2023/05/03 04:34:03 UTC
[shardingsphere] branch master updated: Rename process id to task id in ClusterProcessListSubscriber.postShowProcessListData() (#25437)
This is an automated email from the ASF dual-hosted git repository.
zhangliang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new dd190349365 Rename process id to task id in ClusterProcessListSubscriber.postShowProcessListData() (#25437)
dd190349365 is described below
commit dd19034936585c0d6c828972a3fa462bd488fbda
Author: Liang Zhang <zh...@apache.org>
AuthorDate: Wed May 3 12:33:56 2023 +0800
Rename process id to task id in ClusterProcessListSubscriber.postShowProcessListData() (#25437)
---
.../metadata/persist/node/ComputeNode.java | 8 ++++----
.../metadata/persist/node/ComputeNodeTest.java | 4 ++--
.../subscriber/ClusterProcessListSubscriber.java | 24 +++++++++++-----------
.../subscriber/ProcessListChangedSubscriber.java | 2 +-
4 files changed, 19 insertions(+), 19 deletions(-)
diff --git a/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/node/ComputeNode.java b/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/node/ComputeNode.java
index 0636b850d0e..25dfd4dbd7f 100644
--- a/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/node/ComputeNode.java
+++ b/kernel/metadata/core/src/main/java/org/apache/shardingsphere/metadata/persist/node/ComputeNode.java
@@ -95,14 +95,14 @@ public final class ComputeNode {
}
/**
- * Get process trigger instance process id node path.
+ * Get process trigger instance node path.
*
* @param instanceId instance id
- * @param processId process id
+ * @param taskId show process list task id
* @return path of process trigger instance node path
*/
- public static String getProcessTriggerInstanceIdNodePath(final String instanceId, final String processId) {
- return String.join("/", "", ROOT_NODE, COMPUTE_NODE, PROCESS_TRIGGER, String.join(":", instanceId, processId));
+ public static String getProcessTriggerInstanceNodePath(final String instanceId, final String taskId) {
+ return String.join("/", "", ROOT_NODE, COMPUTE_NODE, PROCESS_TRIGGER, String.join(":", instanceId, taskId));
}
/**
diff --git a/kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/node/ComputeNodeTest.java b/kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/node/ComputeNodeTest.java
index 51a7e7ecfe8..725fc8eb35b 100644
--- a/kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/node/ComputeNodeTest.java
+++ b/kernel/metadata/core/src/test/java/org/apache/shardingsphere/metadata/persist/node/ComputeNodeTest.java
@@ -44,9 +44,9 @@ class ComputeNodeTest {
@Test
void assertGetProcessTriggerInstanceIdNodePath() {
- assertThat(ComputeNode.getProcessTriggerInstanceIdNodePath("foo_instance", "foo_process_id"),
+ assertThat(ComputeNode.getProcessTriggerInstanceNodePath("foo_instance", "foo_process_id"),
is("/nodes/compute_nodes/process_trigger/foo_instance:foo_process_id"));
- assertThat(ComputeNode.getProcessTriggerInstanceIdNodePath("foo_instance", "foo_process_id"),
+ assertThat(ComputeNode.getProcessTriggerInstanceNodePath("foo_instance", "foo_process_id"),
is("/nodes/compute_nodes/process_trigger/foo_instance:foo_process_id"));
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/subscriber/ClusterProcessListSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/subscriber/ClusterProcessListSubscriber.java
index 31bfa25c42e..5062c63a68c 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/subscriber/ClusterProcessListSubscriber.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/registry/process/subscriber/ClusterProcessListSubscriber.java
@@ -55,30 +55,30 @@ public final class ClusterProcessListSubscriber implements ProcessListSubscriber
@Override
@Subscribe
public void postShowProcessListData(final ShowProcessListRequestEvent event) {
- String processId = new UUID(ThreadLocalRandom.current().nextLong(), ThreadLocalRandom.current().nextLong()).toString().replace("-", "");
- boolean triggerIsComplete = false;
- Collection<String> triggerPaths = getTriggerPaths(processId);
+ String taskId = new UUID(ThreadLocalRandom.current().nextLong(), ThreadLocalRandom.current().nextLong()).toString().replace("-", "");
+ boolean isTriggerCompleted = false;
+ Collection<String> triggerPaths = getTriggerPaths(taskId);
try {
triggerPaths.forEach(each -> repository.persist(each, ""));
- triggerIsComplete = waitAllNodeDataReady(processId, triggerPaths);
- postShowProcessListData(processId);
+ isTriggerCompleted = waitAllNodeDataReady(taskId, triggerPaths);
+ postShowProcessListData(taskId);
} finally {
- repository.delete(ProcessNode.getProcessIdPath(processId));
- if (!triggerIsComplete) {
+ repository.delete(ProcessNode.getProcessIdPath(taskId));
+ if (!isTriggerCompleted) {
triggerPaths.forEach(repository::delete);
}
}
}
- private void postShowProcessListData(final String processId) {
- Collection<String> yamlProcessListContexts = repository.getChildrenKeys(ProcessNode.getProcessIdPath(processId)).stream()
- .map(each -> repository.getDirectly(ProcessNode.getProcessListInstancePath(processId, each))).collect(Collectors.toList());
+ private void postShowProcessListData(final String taskId) {
+ Collection<String> yamlProcessListContexts = repository.getChildrenKeys(ProcessNode.getProcessIdPath(taskId)).stream()
+ .map(each -> repository.getDirectly(ProcessNode.getProcessListInstancePath(taskId, each))).collect(Collectors.toList());
eventBusContext.post(new ShowProcessListResponseEvent(yamlProcessListContexts));
}
- private Collection<String> getTriggerPaths(final String processId) {
+ private Collection<String> getTriggerPaths(final String taskId) {
return Stream.of(InstanceType.values())
- .flatMap(each -> repository.getChildrenKeys(ComputeNode.getOnlineNodePath(each)).stream().map(onlinePath -> ComputeNode.getProcessTriggerInstanceIdNodePath(onlinePath, processId)))
+ .flatMap(each -> repository.getChildrenKeys(ComputeNode.getOnlineNodePath(each)).stream().map(onlinePath -> ComputeNode.getProcessTriggerInstanceNodePath(onlinePath, taskId)))
.collect(Collectors.toList());
}
diff --git a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/ProcessListChangedSubscriber.java b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/ProcessListChangedSubscriber.java
index 2db936cb11c..707bbae56d0 100644
--- a/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/ProcessListChangedSubscriber.java
+++ b/mode/type/cluster/core/src/main/java/org/apache/shardingsphere/mode/manager/cluster/coordinator/subscriber/ProcessListChangedSubscriber.java
@@ -69,7 +69,7 @@ public final class ProcessListChangedSubscriber {
registryCenter.getRepository().persist(
ProcessNode.getProcessListInstancePath(event.getProcessId(), event.getInstanceId()), YamlEngine.marshal(swapper.swapToYamlConfiguration(processContexts)));
}
- registryCenter.getRepository().delete(ComputeNode.getProcessTriggerInstanceIdNodePath(event.getInstanceId(), event.getProcessId()));
+ registryCenter.getRepository().delete(ComputeNode.getProcessTriggerInstanceNodePath(event.getInstanceId(), event.getProcessId()));
}
/**