You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by pa...@apache.org on 2022/08/06 11:37:34 UTC
[shardingsphere] branch master updated: Revise job progress inventory and incremental task progress (#19926)
This is an automated email from the ASF dual-hosted git repository.
panjuan 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 7c81a45d2bb Revise job progress inventory and incremental task progress (#19926)
7c81a45d2bb is described below
commit 7c81a45d2bb58d3748b3ec7c0fbb7826a637c724
Author: Hongsheng Zhong <sa...@126.com>
AuthorDate: Sat Aug 6 19:37:28 2022 +0800
Revise job progress inventory and incremental task progress (#19926)
---
...s.java => JobItemIncrementalTasksProgress.java} | 16 +++--
...ess.java => JobItemInventoryTasksProgress.java} | 5 +-
.../pipeline/api/job/progress/JobProgress.java | 16 ++---
...bProgress.java => PipelineJobItemProgress.java} | 4 +-
.../core/api/impl/GovernanceRepositoryAPIImpl.java | 16 ++---
.../yaml/YamlIncrementalTaskProgressSwapper.java | 68 ----------------------
...va => YamlJobItemIncrementalTasksProgress.java} | 4 +-
...YamlJobItemIncrementalTasksProgressSwapper.java | 68 ++++++++++++++++++++++
...java => YamlJobItemInventoryTasksProgress.java} | 4 +-
... YamlJobItemInventoryTasksProgressSwapper.java} | 62 +++++++++-----------
.../core/job/progress/yaml/YamlJobProgress.java | 4 +-
.../job/progress/yaml/YamlJobProgressSwapper.java | 12 ++--
...dleRuleAlteredJobCompletionDetectAlgorithm.java | 4 +-
.../core/job/progress/JobProgressTest.java | 4 +-
...uleAlteredJobCompletionDetectAlgorithmTest.java | 12 ++--
15 files changed, 149 insertions(+), 150 deletions(-)
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobIncrementalTaskProgress.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobItemIncrementalTasksProgress.java
similarity index 83%
rename from shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobIncrementalTaskProgress.java
rename to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobItemIncrementalTasksProgress.java
index 5b6be6dd1cc..ee4b4efeb2d 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobIncrementalTaskProgress.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobItemIncrementalTasksProgress.java
@@ -27,32 +27,30 @@ import org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPositio
import org.apache.shardingsphere.data.pipeline.api.task.progress.IncrementalTaskProgress;
/**
- * Job incrementalTask progress.
+ * Job item incremental tasks progress.
*/
@RequiredArgsConstructor
@Getter
-public class JobIncrementalTaskProgress {
+public final class JobItemIncrementalTasksProgress {
private final Map<String, IncrementalTaskProgress> incrementalTaskProgressMap;
/**
* Get incremental position.
*
- * @param dataSourceName dataSource
+ * @param dataSourceName data source name
* @return incremental position
*/
public Optional<IngestPosition<?>> getIncrementalPosition(final String dataSourceName) {
- Optional<IncrementalTaskProgress> incrementalTaskProgressItem = incrementalTaskProgressMap.entrySet().stream()
- .filter(entry -> dataSourceName.equals(entry.getKey()))
- .map(Map.Entry::getValue)
- .findAny();
- return incrementalTaskProgressItem.map(IncrementalTaskProgress::getPosition);
+ Optional<IncrementalTaskProgress> incrementalTaskProgress = incrementalTaskProgressMap.entrySet().stream()
+ .filter(entry -> dataSourceName.equals(entry.getKey())).map(Map.Entry::getValue).findAny();
+ return incrementalTaskProgress.map(IncrementalTaskProgress::getPosition);
}
/**
* Get incremental latest active time milliseconds.
*
- * @return latest active time, <code>0</code> is there is no activity
+ * @return latest active time, <code>0</code> means there is no activity
*/
public long getIncrementalLatestActiveTimeMillis() {
List<Long> delays = incrementalTaskProgressMap.values().stream()
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobInventoryTaskProgress.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobItemInventoryTasksProgress.java
similarity index 94%
rename from shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobInventoryTaskProgress.java
rename to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobItemInventoryTasksProgress.java
index ff75c1bff5c..b3ff1ad1598 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobInventoryTaskProgress.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobItemInventoryTasksProgress.java
@@ -27,11 +27,11 @@ import org.apache.shardingsphere.data.pipeline.api.ingest.position.IngestPositio
import org.apache.shardingsphere.data.pipeline.api.task.progress.InventoryTaskProgress;
/**
- * Job inventoryTask progress.
+ * Job item inventory tasks progress.
*/
@RequiredArgsConstructor
@Getter
-public class JobInventoryTaskProgress {
+public final class JobItemInventoryTasksProgress {
private final Map<String, InventoryTaskProgress> inventoryTaskProgressMap;
@@ -54,6 +54,7 @@ public class JobInventoryTaskProgress {
* @return finished percentage
*/
public int getInventoryFinishedPercentage() {
+ // TODO finished percentage is not accurate enough
long finished = inventoryTaskProgressMap.values().stream()
.filter(each -> each.getPosition() instanceof FinishedPosition)
.count();
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobProgress.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobProgress.java
index a902fa37884..c210dda4a4e 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobProgress.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/JobProgress.java
@@ -31,7 +31,7 @@ import java.util.Optional;
@Getter
@Setter
// TODO now rename
-public final class JobProgress implements PipelineJobProgress {
+public final class JobProgress implements PipelineJobItemProgress {
private JobStatus status = JobStatus.RUNNING;
@@ -39,9 +39,9 @@ public final class JobProgress implements PipelineJobProgress {
private boolean active;
- private JobInventoryTaskProgress jobInventoryTask;
+ private JobItemInventoryTasksProgress inventory;
- private JobIncrementalTaskProgress jobIncrementalTask;
+ private JobItemIncrementalTasksProgress incremental;
/**
* get incremental position.
@@ -49,7 +49,7 @@ public final class JobProgress implements PipelineJobProgress {
* @return incremental position
*/
public Optional<IngestPosition<?>> getIncrementalPosition(final String dataSourceName) {
- return jobIncrementalTask.getIncrementalPosition(dataSourceName);
+ return incremental.getIncrementalPosition(dataSourceName);
}
/**
@@ -59,7 +59,7 @@ public final class JobProgress implements PipelineJobProgress {
* @return inventory position
*/
public Map<String, IngestPosition<?>> getInventoryPosition(final String tableName) {
- return jobInventoryTask.getInventoryPosition(tableName);
+ return inventory.getInventoryPosition(tableName);
}
/**
@@ -68,7 +68,7 @@ public final class JobProgress implements PipelineJobProgress {
* @return data source
*/
public String getDataSource() {
- return jobIncrementalTask.getIncrementalTaskProgressMap().keySet().stream().findAny().orElse("");
+ return incremental.getIncrementalTaskProgressMap().keySet().stream().findAny().orElse("");
}
/**
@@ -77,7 +77,7 @@ public final class JobProgress implements PipelineJobProgress {
* @return finished percentage
*/
public int getInventoryFinishedPercentage() {
- return jobInventoryTask.getInventoryFinishedPercentage();
+ return inventory.getInventoryFinishedPercentage();
}
/**
@@ -86,6 +86,6 @@ public final class JobProgress implements PipelineJobProgress {
* @return latest active time, <code>0</code> is there is no activity
*/
public long getIncrementalLatestActiveTimeMillis() {
- return null == jobIncrementalTask ? 0L : jobIncrementalTask.getIncrementalLatestActiveTimeMillis();
+ return null == incremental ? 0L : incremental.getIncrementalLatestActiveTimeMillis();
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/PipelineJobProgress.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/PipelineJobItemProgress.java
similarity index 92%
rename from shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/PipelineJobProgress.java
rename to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/PipelineJobItemProgress.java
index ab7a8e080bb..501e4100534 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/PipelineJobProgress.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-api/src/main/java/org/apache/shardingsphere/data/pipeline/api/job/progress/PipelineJobItemProgress.java
@@ -18,7 +18,7 @@
package org.apache.shardingsphere.data.pipeline.api.job.progress;
/**
- * Pipeline job progress.
+ * Pipeline job item progress.
*/
-public interface PipelineJobProgress {
+public interface PipelineJobItemProgress {
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/GovernanceRepositoryAPIImpl.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/GovernanceRepositoryAPIImpl.java
index 06c5a61a25a..4f6f3461d18 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/GovernanceRepositoryAPIImpl.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/api/impl/GovernanceRepositoryAPIImpl.java
@@ -21,8 +21,8 @@ import com.google.common.base.Strings;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.shardingsphere.data.pipeline.api.job.JobStatus;
-import org.apache.shardingsphere.data.pipeline.api.job.progress.JobIncrementalTaskProgress;
-import org.apache.shardingsphere.data.pipeline.api.job.progress.JobInventoryTaskProgress;
+import org.apache.shardingsphere.data.pipeline.api.job.progress.JobItemIncrementalTasksProgress;
+import org.apache.shardingsphere.data.pipeline.api.job.progress.JobItemInventoryTasksProgress;
import org.apache.shardingsphere.data.pipeline.api.job.progress.JobProgress;
import org.apache.shardingsphere.data.pipeline.core.api.GovernanceRepositoryAPI;
import org.apache.shardingsphere.data.pipeline.core.job.progress.yaml.YamlJobProgressSwapper;
@@ -60,20 +60,20 @@ public final class GovernanceRepositoryAPIImpl implements GovernanceRepositoryAP
JobProgress jobProgress = new JobProgress();
jobProgress.setStatus(jobContext.getStatus());
jobProgress.setSourceDatabaseType(jobContext.getJobConfig().getSourceDatabaseType());
- jobProgress.setJobIncrementalTask(getIncrementalTaskProgress(jobContext));
- jobProgress.setJobInventoryTask(getInventoryTaskProgress(jobContext));
+ jobProgress.setIncremental(getIncrementalTasksProgress(jobContext));
+ jobProgress.setInventory(getInventoryTasksProgress(jobContext));
String value = YamlEngine.marshal(SWAPPER.swapToYaml(jobProgress));
repository.persist(PipelineMetaDataNode.getScalingJobOffsetPath(jobContext.getJobId(), jobContext.getShardingItem()), value);
}
- private JobIncrementalTaskProgress getIncrementalTaskProgress(final RuleAlteredJobContext jobContext) {
- return new JobIncrementalTaskProgress(
+ private JobItemIncrementalTasksProgress getIncrementalTasksProgress(final RuleAlteredJobContext jobContext) {
+ return new JobItemIncrementalTasksProgress(
jobContext.getIncrementalTasks()
.stream().collect(Collectors.toMap(IncrementalTask::getTaskId, IncrementalTask::getProgress)));
}
- private JobInventoryTaskProgress getInventoryTaskProgress(final RuleAlteredJobContext jobContext) {
- return new JobInventoryTaskProgress(
+ private JobItemInventoryTasksProgress getInventoryTasksProgress(final RuleAlteredJobContext jobContext) {
+ return new JobItemInventoryTasksProgress(
jobContext.getInventoryTasks()
.stream()
.collect(Collectors.toMap(InventoryTask::getTaskId, InventoryTask::getProgress)));
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlIncrementalTaskProgressSwapper.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlIncrementalTaskProgressSwapper.java
deleted file mode 100644
index 8c7376fda7d..00000000000
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlIncrementalTaskProgressSwapper.java
+++ /dev/null
@@ -1,68 +0,0 @@
-/*
- * 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.data.pipeline.core.job.progress.yaml;
-
-import java.util.Collections;
-import org.apache.shardingsphere.data.pipeline.api.job.progress.JobIncrementalTaskProgress;
-import org.apache.shardingsphere.data.pipeline.api.task.progress.IncrementalTaskProgress;
-import org.apache.shardingsphere.data.pipeline.core.ingest.position.PositionInitializerFactory;
-
-/**
- * YAML IncrementalTask progress swapper.
- */
-public final class YamlIncrementalTaskProgressSwapper {
-
- /**
- * Swap to YAML.
- *
- * @param jobIncrementalTask incrementalTask progress
- * @return IncrementalTaskProgress
- */
- public YamlIncrementalTaskProgress swapToYaml(final JobIncrementalTaskProgress jobIncrementalTask) {
- if (null == jobIncrementalTask) {
- return null;
- }
- return jobIncrementalTask.getIncrementalTaskProgressMap()
- .entrySet()
- .stream()
- .map(entry -> {
- YamlIncrementalTaskProgress yamlIncrementalTaskProgress = new YamlIncrementalTaskProgress();
- yamlIncrementalTaskProgress.setDataSourceName(entry.getKey());
- yamlIncrementalTaskProgress.setPosition(entry.getValue().getPosition().toString());
- yamlIncrementalTaskProgress.setDelay(entry.getValue().getIncrementalTaskDelay());
- return yamlIncrementalTaskProgress;
- }).findAny().orElse(null);
- }
-
- /**
- * Swap to object.
- *
- * @param databaseType databaseType
- * @param incremental yaml incrementalTask progress
- * @return incrementalTask progress
- */
- public JobIncrementalTaskProgress swapToObject(final String databaseType, final YamlIncrementalTaskProgress incremental) {
- if (null == incremental) {
- return null;
- }
- IncrementalTaskProgress result = new IncrementalTaskProgress();
- result.setPosition(PositionInitializerFactory.getInstance(databaseType).init(incremental.getPosition()));
- result.setIncrementalTaskDelay(incremental.getDelay());
- return new JobIncrementalTaskProgress(Collections.singletonMap(incremental.getDataSourceName(), result));
- }
-}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlIncrementalTaskProgress.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemIncrementalTasksProgress.java
similarity index 91%
rename from shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlIncrementalTaskProgress.java
rename to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemIncrementalTasksProgress.java
index 431b69c4e08..c7070c10a3f 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlIncrementalTaskProgress.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemIncrementalTasksProgress.java
@@ -22,11 +22,11 @@ import lombok.Setter;
import org.apache.shardingsphere.data.pipeline.api.task.progress.IncrementalTaskDelay;
/**
- * Yaml IncrementalTaskProgress.
+ * YAML job item incremental tasks progress.
*/
@Getter
@Setter
-public final class YamlIncrementalTaskProgress {
+public final class YamlJobItemIncrementalTasksProgress {
private String dataSourceName;
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemIncrementalTasksProgressSwapper.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemIncrementalTasksProgressSwapper.java
new file mode 100644
index 00000000000..c2d7a467b53
--- /dev/null
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemIncrementalTasksProgressSwapper.java
@@ -0,0 +1,68 @@
+/*
+ * 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.data.pipeline.core.job.progress.yaml;
+
+import java.util.Collections;
+import org.apache.shardingsphere.data.pipeline.api.job.progress.JobItemIncrementalTasksProgress;
+import org.apache.shardingsphere.data.pipeline.api.task.progress.IncrementalTaskProgress;
+import org.apache.shardingsphere.data.pipeline.core.ingest.position.PositionInitializerFactory;
+
+/**
+ * YAML job item incremental tasks progress swapper.
+ */
+public final class YamlJobItemIncrementalTasksProgressSwapper {
+
+ /**
+ * Swap to YAML.
+ *
+ * @param progress progress
+ * @return YAML progress
+ */
+ public YamlJobItemIncrementalTasksProgress swapToYaml(final JobItemIncrementalTasksProgress progress) {
+ if (null == progress) {
+ return null;
+ }
+ return progress.getIncrementalTaskProgressMap()
+ .entrySet().stream()
+ .map(entry -> {
+ YamlJobItemIncrementalTasksProgress result = new YamlJobItemIncrementalTasksProgress();
+ result.setDataSourceName(entry.getKey());
+ result.setPosition(entry.getValue().getPosition().toString());
+ result.setDelay(entry.getValue().getIncrementalTaskDelay());
+ return result;
+ }).findAny().orElse(null);
+ }
+
+ /**
+ * Swap to object.
+ *
+ * @param databaseType database type
+ * @param yamlProgress YAML progress
+ * @return progress
+ */
+ public JobItemIncrementalTasksProgress swapToObject(final String databaseType, final YamlJobItemIncrementalTasksProgress yamlProgress) {
+ if (null == yamlProgress) {
+ return null;
+ }
+ IncrementalTaskProgress taskProgress = new IncrementalTaskProgress();
+ // TODO databaseType
+ taskProgress.setPosition(PositionInitializerFactory.getInstance(databaseType).init(yamlProgress.getPosition()));
+ taskProgress.setIncrementalTaskDelay(yamlProgress.getDelay());
+ return new JobItemIncrementalTasksProgress(Collections.singletonMap(yamlProgress.getDataSourceName(), taskProgress));
+ }
+}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlInventoryTaskProgress.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemInventoryTasksProgress.java
similarity index 91%
rename from shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlInventoryTaskProgress.java
rename to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemInventoryTasksProgress.java
index 9aefc9ab639..ea7a651f18f 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlInventoryTaskProgress.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemInventoryTasksProgress.java
@@ -23,11 +23,11 @@ import lombok.Getter;
import lombok.Setter;
/**
- * Yaml InventoryTaskProgress.
+ * YAML job item inventory tasks progress.
*/
@Getter
@Setter
-public final class YamlInventoryTaskProgress {
+public final class YamlJobItemInventoryTasksProgress {
private String[] finished = new String[0];
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlInventoryTaskProgressSwapper.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemInventoryTasksProgressSwapper.java
similarity index 55%
rename from shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlInventoryTaskProgressSwapper.java
rename to shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemInventoryTasksProgressSwapper.java
index d5578437f40..2516af04b6a 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlInventoryTaskProgressSwapper.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobItemInventoryTasksProgressSwapper.java
@@ -18,46 +18,46 @@
package org.apache.shardingsphere.data.pipeline.core.job.progress.yaml;
import com.google.common.base.Strings;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.function.Function;
-import java.util.stream.Collectors;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.FinishedPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PlaceholderPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PrimaryKeyPositionFactory;
-import org.apache.shardingsphere.data.pipeline.api.job.progress.JobInventoryTaskProgress;
+import org.apache.shardingsphere.data.pipeline.api.job.progress.JobItemInventoryTasksProgress;
import org.apache.shardingsphere.data.pipeline.api.task.progress.InventoryTaskProgress;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
/**
- * YAML InventoryTask progress swapper.
+ * YAML job item inventory tasks progress swapper.
*/
-public final class YamlInventoryTaskProgressSwapper {
+public final class YamlJobItemInventoryTasksProgressSwapper {
/**
* Swap to YAML.
*
- * @param inventory inventoryTask progress
- * @return YAML inventoryTask progress
+ * @param progress progress
+ * @return YAML progress
*/
- public YamlInventoryTaskProgress swapToYaml(final JobInventoryTaskProgress inventory) {
- YamlInventoryTaskProgress result = new YamlInventoryTaskProgress();
- if (inventory != null) {
- result.setFinished(getFinished(inventory));
- result.setUnfinished(getUnfinished(inventory));
+ public YamlJobItemInventoryTasksProgress swapToYaml(final JobItemInventoryTasksProgress progress) {
+ YamlJobItemInventoryTasksProgress result = new YamlJobItemInventoryTasksProgress();
+ if (progress != null) {
+ result.setFinished(getFinished(progress));
+ result.setUnfinished(getUnfinished(progress));
}
return result;
}
- private String[] getFinished(final JobInventoryTaskProgress jobInventoryTaskProgress) {
- return jobInventoryTaskProgress.getInventoryTaskProgressMap().entrySet().stream()
+ private String[] getFinished(final JobItemInventoryTasksProgress progress) {
+ return progress.getInventoryTaskProgressMap().entrySet().stream()
.filter(entry -> entry.getValue().getPosition() instanceof FinishedPosition)
- .map(Map.Entry::getKey)
- .toArray(String[]::new);
+ .map(Map.Entry::getKey).toArray(String[]::new);
}
- private Map<String, String> getUnfinished(final JobInventoryTaskProgress jobInventoryTaskProgress) {
- return jobInventoryTaskProgress.getInventoryTaskProgressMap().entrySet().stream()
+ private Map<String, String> getUnfinished(final JobItemInventoryTasksProgress progress) {
+ return progress.getInventoryTaskProgressMap().entrySet().stream()
.filter(entry -> !(entry.getValue().getPosition() instanceof FinishedPosition))
.collect(Collectors.toMap(Map.Entry::getKey, entry -> entry.getValue().getPosition().toString()));
}
@@ -65,21 +65,17 @@ public final class YamlInventoryTaskProgressSwapper {
/**
* Swap to object.
*
- * @param inventory yaml inventoryTask progress
- * @return inventoryTask progress
+ * @param yamlProgress YAML progress
+ * @return progress
*/
- public JobInventoryTaskProgress swapToObject(final YamlInventoryTaskProgress inventory) {
- if (null == inventory) {
+ public JobItemInventoryTasksProgress swapToObject(final YamlJobItemInventoryTasksProgress yamlProgress) {
+ if (null == yamlProgress) {
return null;
}
- Map<String, InventoryTaskProgress> inventoryTaskProgressItemMap = new HashMap<>();
- inventoryTaskProgressItemMap.putAll(Arrays.stream(inventory.getFinished())
- .collect(Collectors.toMap(key -> key, value -> new InventoryTaskProgress(new FinishedPosition()))));
- inventoryTaskProgressItemMap.putAll(inventory.getUnfinished().entrySet()
- .stream()
- .collect(Collectors.toMap(
- Map.Entry::getKey, getInventoryTaskProgressFunction())));
- return new JobInventoryTaskProgress(inventoryTaskProgressItemMap);
+ Map<String, InventoryTaskProgress> taskProgressMap = new LinkedHashMap<>();
+ taskProgressMap.putAll(Arrays.stream(yamlProgress.getFinished()).collect(Collectors.toMap(key -> key, value -> new InventoryTaskProgress(new FinishedPosition()))));
+ taskProgressMap.putAll(yamlProgress.getUnfinished().entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, getInventoryTaskProgressFunction())));
+ return new JobItemInventoryTasksProgress(taskProgressMap);
}
private Function<Map.Entry<String, String>, InventoryTaskProgress> getInventoryTaskProgressFunction() {
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobProgress.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobProgress.java
index 6f060f6ba1d..a5be79669b7 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobProgress.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobProgress.java
@@ -31,7 +31,7 @@ public final class YamlJobProgress {
private String sourceDatabaseType;
- private YamlInventoryTaskProgress inventory;
+ private YamlJobItemInventoryTasksProgress inventory;
- private YamlIncrementalTaskProgress incremental;
+ private YamlJobItemIncrementalTasksProgress incremental;
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobProgressSwapper.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobProgressSwapper.java
index 6b0387ab6d3..2cec5b1d453 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobProgressSwapper.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/yaml/YamlJobProgressSwapper.java
@@ -25,6 +25,10 @@ import org.apache.shardingsphere.data.pipeline.api.job.progress.JobProgress;
*/
public final class YamlJobProgressSwapper {
+ private static final YamlJobItemInventoryTasksProgressSwapper INVENTORY_PROGRESS_SWAPPER = new YamlJobItemInventoryTasksProgressSwapper();
+
+ private static final YamlJobItemIncrementalTasksProgressSwapper INCREMENTAL_PROGRESS_SWAPPER = new YamlJobItemIncrementalTasksProgressSwapper();
+
/**
* Swap to YAML.
*
@@ -35,8 +39,8 @@ public final class YamlJobProgressSwapper {
YamlJobProgress result = new YamlJobProgress();
result.setStatus(jobProgress.getStatus().name());
result.setSourceDatabaseType(jobProgress.getSourceDatabaseType());
- result.setInventory(new YamlInventoryTaskProgressSwapper().swapToYaml(jobProgress.getJobInventoryTask()));
- result.setIncremental(new YamlIncrementalTaskProgressSwapper().swapToYaml(jobProgress.getJobIncrementalTask()));
+ result.setInventory(INVENTORY_PROGRESS_SWAPPER.swapToYaml(jobProgress.getInventory()));
+ result.setIncremental(INCREMENTAL_PROGRESS_SWAPPER.swapToYaml(jobProgress.getIncremental()));
return result;
}
@@ -50,8 +54,8 @@ public final class YamlJobProgressSwapper {
JobProgress result = new JobProgress();
result.setStatus(JobStatus.valueOf(yamlJobProgress.getStatus()));
result.setSourceDatabaseType(yamlJobProgress.getSourceDatabaseType());
- result.setJobInventoryTask(new YamlInventoryTaskProgressSwapper().swapToObject(yamlJobProgress.getInventory()));
- result.setJobIncrementalTask(new YamlIncrementalTaskProgressSwapper().swapToObject(yamlJobProgress.getSourceDatabaseType(), yamlJobProgress.getIncremental()));
+ result.setInventory(INVENTORY_PROGRESS_SWAPPER.swapToObject(yamlJobProgress.getInventory()));
+ result.setIncremental(INCREMENTAL_PROGRESS_SWAPPER.swapToObject(yamlJobProgress.getSourceDatabaseType(), yamlJobProgress.getIncremental()));
return result;
}
}
diff --git a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/spi/IdleRuleAlteredJobCompletionDetectAlgorithm.java b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/spi/IdleRuleAlteredJobCompletionDetectAlgorithm.java
index dea0d03c684..e0d422da6b5 100644
--- a/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/spi/IdleRuleAlteredJobCompletionDetectAlgorithm.java
+++ b/shardingsphere-kernel/shardingsphere-data-pipeline/shardingsphere-data-pipeline-core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/spi/IdleRuleAlteredJobCompletionDetectAlgorithm.java
@@ -75,12 +75,12 @@ public final class IdleRuleAlteredJobCompletionDetectAlgorithm implements JobCom
}
private static boolean isAllInventoryTasksCompleted(final Collection<JobProgress> jobProgresses) {
- return jobProgresses.stream().flatMap(each -> each.getJobInventoryTask().getInventoryTaskProgressMap().values().stream()).allMatch(each -> each.getPosition() instanceof FinishedPosition);
+ return jobProgresses.stream().flatMap(each -> each.getInventory().getInventoryTaskProgressMap().values().stream()).allMatch(each -> each.getPosition() instanceof FinishedPosition);
}
private static Collection<Long> getIncrementalTasksIdleSeconds(final Collection<JobProgress> jobProgresses) {
long currentTimeMillis = System.currentTimeMillis();
- return jobProgresses.stream().flatMap(each -> each.getJobIncrementalTask().getIncrementalTaskProgressMap().values().stream())
+ return jobProgresses.stream().flatMap(each -> each.getIncremental().getIncrementalTaskProgressMap().values().stream())
.map(each -> {
long latestActiveTimeMillis = each.getIncrementalTaskDelay().getLatestActiveTimeMillis();
return latestActiveTimeMillis > 0 ? TimeUnit.MILLISECONDS.toSeconds(currentTimeMillis - latestActiveTimeMillis) : 0;
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
index 695258f4d28..1fc6afb8f4a 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/core/job/progress/JobProgressTest.java
@@ -45,8 +45,8 @@ public final class JobProgressTest {
JobProgress jobProgress = getJobProgress(ConfigurationFileUtil.readFile("job-progress.yaml"));
assertThat(jobProgress.getStatus(), is(JobStatus.RUNNING));
assertThat(jobProgress.getSourceDatabaseType(), is("H2"));
- assertThat(jobProgress.getJobInventoryTask().getInventoryTaskProgressMap().size(), is(4));
- assertThat(jobProgress.getJobIncrementalTask().getIncrementalTaskProgressMap().size(), is(1));
+ assertThat(jobProgress.getInventory().getInventoryTaskProgressMap().size(), is(4));
+ assertThat(jobProgress.getIncremental().getIncrementalTaskProgressMap().size(), is(1));
}
@Test
diff --git a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/IdleRuleAlteredJobCompletionDetectAlgorithmTest.java b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/IdleRuleAlteredJobCompletionDetectAlgorithmTest.java
index 141abac58e0..786b5c262d2 100644
--- a/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/IdleRuleAlteredJobCompletionDetectAlgorithmTest.java
+++ b/shardingsphere-test/shardingsphere-pipeline-test/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/rulealtered/IdleRuleAlteredJobCompletionDetectAlgorithmTest.java
@@ -20,8 +20,8 @@ package org.apache.shardingsphere.data.pipeline.scenario.rulealtered;
import org.apache.shardingsphere.data.pipeline.api.detect.RuleAlteredJobAlmostCompletedParameter;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.FinishedPosition;
import org.apache.shardingsphere.data.pipeline.api.ingest.position.PlaceholderPosition;
-import org.apache.shardingsphere.data.pipeline.api.job.progress.JobIncrementalTaskProgress;
-import org.apache.shardingsphere.data.pipeline.api.job.progress.JobInventoryTaskProgress;
+import org.apache.shardingsphere.data.pipeline.api.job.progress.JobItemIncrementalTasksProgress;
+import org.apache.shardingsphere.data.pipeline.api.job.progress.JobItemInventoryTasksProgress;
import org.apache.shardingsphere.data.pipeline.api.job.progress.JobProgress;
import org.apache.shardingsphere.data.pipeline.api.task.progress.IncrementalTaskProgress;
import org.apache.shardingsphere.data.pipeline.api.task.progress.InventoryTaskProgress;
@@ -68,7 +68,7 @@ public final class IdleRuleAlteredJobCompletionDetectAlgorithmTest {
public void assertFalseOnUnFinishedPosition() {
int jobShardingCount = 1;
JobProgress jobProgress = new JobProgress();
- jobProgress.setJobInventoryTask(new JobInventoryTaskProgress(Collections.singletonMap("foo_ds", new InventoryTaskProgress(new PlaceholderPosition()))));
+ jobProgress.setInventory(new JobItemInventoryTasksProgress(Collections.singletonMap("foo_ds", new InventoryTaskProgress(new PlaceholderPosition()))));
Collection<JobProgress> jobProgresses = Collections.singleton(jobProgress);
RuleAlteredJobAlmostCompletedParameter parameter = new RuleAlteredJobAlmostCompletedParameter(jobShardingCount, jobProgresses);
assertFalse(JobCompletionDetectAlgorithmFactory.newInstance(new AlgorithmConfiguration("IDLE", new Properties())).isAlmostCompleted(parameter));
@@ -98,11 +98,11 @@ public final class IdleRuleAlteredJobCompletionDetectAlgorithmTest {
private JobProgress createJobProgress(final long latestActiveTimeMillis) {
JobProgress result = new JobProgress();
- result.setJobInventoryTask(new JobInventoryTaskProgress(Collections.singletonMap("foo_ds", new InventoryTaskProgress(new FinishedPosition()))));
+ result.setInventory(new JobItemInventoryTasksProgress(Collections.singletonMap("foo_ds", new InventoryTaskProgress(new FinishedPosition()))));
IncrementalTaskProgress incrementalTaskProgress = new IncrementalTaskProgress();
incrementalTaskProgress.getIncrementalTaskDelay().setLatestActiveTimeMillis(latestActiveTimeMillis);
- JobIncrementalTaskProgress jobIncrementalTaskProgress = new JobIncrementalTaskProgress(Collections.singletonMap("foo_ds", incrementalTaskProgress));
- result.setJobIncrementalTask(jobIncrementalTaskProgress);
+ JobItemIncrementalTasksProgress incremental = new JobItemIncrementalTasksProgress(Collections.singletonMap("foo_ds", incrementalTaskProgress));
+ result.setIncremental(incremental);
return result;
}
}