This is an automated email from the ASF dual-hosted git repository.
terrymanu 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 5cec93484b4 Persist consistency check job phase status (#39816)
5cec93484b4 is described below
commit 5cec93484b445d4fc64b02b15de874d83a2b6c09
Author: Liang Zhang <[email protected]>
AuthorDate: Sat Sep 12 21:54:03 2026 +0800
Persist consistency check job phase status (#39816)
* Persist consistency check job phase status
* Persist consistency check job phase status
---
...tractDelegatingDialectDatabaseMetaDataTest.java | 6 --
.../migration/show-migration-check-status.cn.md | 10 +--
.../migration/show-migration-check-status.en.md | 10 +--
.../ConsistencyCheckJobItemProgressContext.java | 3 +
.../pojo/ConsistencyCheckJobItemInfo.java | 3 +
.../progress/ConsistencyCheckJobItemProgress.java | 1 +
.../api/ConsistencyCheckJobAPI.java | 1 +
.../context/ConsistencyCheckJobItemContext.java | 18 +++--
.../ConsistencyCheckJobItemContextTest.java | 12 +++
.../MigrationDataConsistencyChecker.java | 4 +
.../query/ShowMigrationCheckStatusExecutor.java | 6 +-
.../ShowMigrationCheckStatusExecutorTest.java | 91 ++++++++++++++++++++++
.../api/ConsistencyCheckJobAPITest.java | 1 +
.../MigrationDataConsistencyCheckerTest.java | 12 +--
14 files changed, 148 insertions(+), 30 deletions(-)
diff --git
a/database/connector/core/src/test/java/org/apache/shardingsphere/database/connector/core/metadata/database/metadata/AbstractDelegatingDialectDatabaseMetaDataTest.java
b/database/connector/core/src/test/java/org/apache/shardingsphere/database/connector/core/metadata/database/metadata/AbstractDelegatingDialectDatabaseMetaDataTest.java
index 8f0fff25748..b077a6da719 100644
---
a/database/connector/core/src/test/java/org/apache/shardingsphere/database/connector/core/metadata/database/metadata/AbstractDelegatingDialectDatabaseMetaDataTest.java
+++
b/database/connector/core/src/test/java/org/apache/shardingsphere/database/connector/core/metadata/database/metadata/AbstractDelegatingDialectDatabaseMetaDataTest.java
@@ -41,7 +41,6 @@ import java.util.Optional;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
-import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.when;
@@ -92,11 +91,6 @@ class AbstractDelegatingDialectDatabaseMetaDataTest {
assertThat(metaData.getFunctionOption(),
is(delegate.getFunctionOption()));
}
- @Test
- void assertRejectNullDelegate() {
- assertThrows(NullPointerException.class, () -> new
FixtureDelegatingDialectDatabaseMetaData(null));
- }
-
@Test
void assertBranchIdentity() {
assertThat(metaData.getDatabaseType(), is("BRANCH"));
diff --git
a/docs/document/content/user-manual/shardingsphere-proxy/distsql/syntax/ral/migration/show-migration-check-status.cn.md
b/docs/document/content/user-manual/shardingsphere-proxy/distsql/syntax/ral/migration/show-migration-check-status.cn.md
index 256373ec668..a8f05bb6c62 100644
---
a/docs/document/content/user-manual/shardingsphere-proxy/distsql/syntax/ral/migration/show-migration-check-status.cn.md
+++
b/docs/document/content/user-manual/shardingsphere-proxy/distsql/syntax/ral/migration/show-migration-check-status.cn.md
@@ -57,11 +57,11 @@ SHOW MIGRATION CHECK STATUS
'j010180026753ef0e25d3932d94d1673ba551';
```sql
mysql> SHOW MIGRATION CHECK STATUS 'j010180026753ef0e25d3932d94d1673ba551';
-+---------+--------+---------------------+----------------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
-| tables | result | check_failed_tables | ignored_tables | active |
inventory_finished_percentage | inventory_remaining_seconds |
incremental_idle_seconds | check_begin_time | check_end_time |
duration_seconds | algorithm_type | algorithm_props | error_message |
-+---------+--------+---------------------+----------------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
-| t_order | true | | | false | 100
| 0 | 0 |
2022-11-01 17:57:39.940 | 2022-11-01 17:57:40.587 | 0 |
DATA_MATCH | | |
-+---------+--------+---------------------+----------------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
++---------+--------+---------------------+----------------+----------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
+| tables | result | check_failed_tables | ignored_tables | status | active
| inventory_finished_percentage | inventory_remaining_seconds |
incremental_idle_seconds | check_begin_time | check_end_time |
duration_seconds | algorithm_type | algorithm_props | error_message |
++---------+--------+---------------------+----------------+----------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
+| t_order | true | | | FINISHED | false
| 100 | 0 | 0
| 2022-11-01 17:57:39.940 | 2022-11-01 17:57:40.587 | 0
| DATA_MATCH | | |
++---------+--------+---------------------+----------------+----------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
1 row in set (0.01 sec)
```
diff --git
a/docs/document/content/user-manual/shardingsphere-proxy/distsql/syntax/ral/migration/show-migration-check-status.en.md
b/docs/document/content/user-manual/shardingsphere-proxy/distsql/syntax/ral/migration/show-migration-check-status.en.md
index 22a64479e03..92c251169fc 100644
---
a/docs/document/content/user-manual/shardingsphere-proxy/distsql/syntax/ral/migration/show-migration-check-status.en.md
+++
b/docs/document/content/user-manual/shardingsphere-proxy/distsql/syntax/ral/migration/show-migration-check-status.en.md
@@ -57,11 +57,11 @@ SHOW MIGRATION CHECK STATUS
'j010180026753ef0e25d3932d94d1673ba551';
```sql
mysql> SHOW MIGRATION CHECK STATUS 'j010180026753ef0e25d3932d94d1673ba551';
-+---------+--------+---------------------+----------------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
-| tables | result | check_failed_tables | ignored_tables | active |
inventory_finished_percentage | inventory_remaining_seconds |
incremental_idle_seconds | check_begin_time | check_end_time |
duration_seconds | algorithm_type | algorithm_props | error_message |
-+---------+--------+---------------------+----------------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
-| t_order | true | | | false | 100
| 0 | 0 |
2022-11-01 17:57:39.940 | 2022-11-01 17:57:40.587 | 0 |
DATA_MATCH | | |
-+---------+--------+---------------------+----------------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
++---------+--------+---------------------+----------------+----------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
+| tables | result | check_failed_tables | ignored_tables | status | active
| inventory_finished_percentage | inventory_remaining_seconds |
incremental_idle_seconds | check_begin_time | check_end_time |
duration_seconds | algorithm_type | algorithm_props | error_message |
++---------+--------+---------------------+----------------+----------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
+| t_order | true | | | FINISHED | false
| 100 | 0 | 0
| 2022-11-01 17:57:39.940 | 2022-11-01 17:57:40.587 | 0
| DATA_MATCH | | |
++---------+--------+---------------------+----------------+----------+--------+-------------------------------+-----------------------------+--------------------------+-------------------------+-------------------------+------------------+----------------+-----------------+---------------+
1 row in set (0.01 sec)
```
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/ConsistencyCheckJobItemProgressContext.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/ConsistencyCheckJobItemProgressContext.java
index cfd2810435b..6a5a09c575a 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/ConsistencyCheckJobItemProgressContext.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/ConsistencyCheckJobItemProgressContext.java
@@ -21,6 +21,7 @@ import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.Setter;
import
org.apache.shardingsphere.data.pipeline.core.consistencycheck.position.TableCheckRangePosition;
+import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
import
org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobProgressListener;
import
org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobUpdateProgress;
import
org.apache.shardingsphere.data.pipeline.core.job.progress.persist.PipelineJobProgressPersistService;
@@ -59,6 +60,8 @@ public final class ConsistencyCheckJobItemProgressContext
implements PipelineJob
private final String sourceDatabaseType;
+ private volatile JobStatus status;
+
@Override
public void onProgressUpdated(final PipelineJobUpdateProgress
updateProgress) {
checkedRecordsCount.addAndGet(updateProgress.getProcessedRecordsCount());
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/pojo/ConsistencyCheckJobItemInfo.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/pojo/ConsistencyCheckJobItemInfo.java
index 1c163ecfe4e..a0b6fefaca1 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/pojo/ConsistencyCheckJobItemInfo.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/pojo/ConsistencyCheckJobItemInfo.java
@@ -19,6 +19,7 @@ package
org.apache.shardingsphere.data.pipeline.core.consistencycheck.pojo;
import lombok.Getter;
import lombok.Setter;
+import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
/**
* Consistency check job item info.
@@ -31,6 +32,8 @@ public final class ConsistencyCheckJobItemInfo {
private boolean active;
+ private JobStatus status;
+
private String tableNames;
private Boolean checkSuccess;
diff --git
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/ConsistencyCheckJobItemProgress.java
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/ConsistencyCheckJobItemProgress.java
index e21b7c3a711..5cc12135f3c 100644
---
a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/ConsistencyCheckJobItemProgress.java
+++
b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/job/progress/ConsistencyCheckJobItemProgress.java
@@ -65,5 +65,6 @@ public final class ConsistencyCheckJobItemProgress implements
PipelineJobItemPro
checkEndTimeMillis = context.getCheckEndTimeMillis();
tableCheckRangePositions.addAll(context.getTableCheckRangePositions());
sourceDatabaseType = context.getSourceDatabaseType();
+ status = context.getStatus();
}
}
diff --git
a/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/ConsistencyCheckJobAPI.java
b/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/ConsistencyCheckJobAPI.java
index 9cb07e72750..66cbcc8592e 100644
---
a/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/ConsistencyCheckJobAPI.java
+++
b/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/ConsistencyCheckJobAPI.java
@@ -176,6 +176,7 @@ public final class ConsistencyCheckJobAPI {
return result;
}
ConsistencyCheckJobItemProgress jobItemProgress = progress.get();
+ result.setStatus(jobItemProgress.getStatus());
if (null == jobItemProgress.getRecordsCount() || null ==
jobItemProgress.getCheckedRecordsCount()) {
result.setInventoryFinishedPercentage(0);
result.setCheckSuccess(null);
diff --git
a/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContext.java
b/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContext.java
index bb56b31df97..8b62834ba35 100644
---
a/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContext.java
+++
b/kernel/data-pipeline/scenario/consistency-check/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContext.java
@@ -43,8 +43,6 @@ public final class ConsistencyCheckJobItemContext implements
PipelineJobItemCont
private volatile boolean stopping;
- private volatile JobStatus status;
-
private final ConsistencyCheckJobConfiguration jobConfig;
private final ConsistencyCheckJobItemProgressContext progressContext;
@@ -55,8 +53,8 @@ public final class ConsistencyCheckJobItemContext implements
PipelineJobItemCont
this.jobConfig = jobConfig;
jobId = jobConfig.getJobId();
this.shardingItem = shardingItem;
- this.status = status;
progressContext = new ConsistencyCheckJobItemProgressContext(jobId,
shardingItem, jobConfig.getSourceDatabaseType().getType());
+ progressContext.setStatus(status);
if (null != jobItemProgress) {
progressContext.getCheckedRecordsCount().set(Optional.ofNullable(jobItemProgress.getCheckedRecordsCount()).orElse(0L));
progressContext.getTableCheckRangePositions().addAll(jobItemProgress.getTableCheckRangePositions());
@@ -64,6 +62,16 @@ public final class ConsistencyCheckJobItemContext implements
PipelineJobItemCont
processContext = new ConsistencyCheckProcessContext(jobId);
}
+ @Override
+ public JobStatus getStatus() {
+ return progressContext.getStatus();
+ }
+
+ @Override
+ public void setStatus(final JobStatus status) {
+ progressContext.setStatus(status);
+ }
+
@Override
public PipelineProcessContext getJobProcessContext() {
return processContext;
@@ -71,8 +79,6 @@ public final class ConsistencyCheckJobItemContext implements
PipelineJobItemCont
@Override
public ConsistencyCheckJobItemProgress toProgress() {
- ConsistencyCheckJobItemProgress result = new
ConsistencyCheckJobItemProgress(progressContext);
- result.setStatus(status);
- return result;
+ return new ConsistencyCheckJobItemProgress(progressContext);
}
}
diff --git
a/kernel/data-pipeline/scenario/consistency-check/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContextTest.java
b/kernel/data-pipeline/scenario/consistency-check/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContextTest.java
index a2814ef01ca..04bbc121652 100644
---
a/kernel/data-pipeline/scenario/consistency-check/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContextTest.java
+++
b/kernel/data-pipeline/scenario/consistency-check/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContextTest.java
@@ -86,4 +86,16 @@ class ConsistencyCheckJobItemContextTest {
assertThat(actual.getLowerBound(), is(expected.getLowerBound()));
assertThat(actual.getUpperBound(), is(expected.getUpperBound()));
}
+
+ @Test
+ void assertStatus() {
+ ConsistencyCheckJobItemContext actual = new
ConsistencyCheckJobItemContext(
+ new ConsistencyCheckJobConfiguration("", "", "DATA_MATCH",
null, databaseType), 0, JobStatus.PREPARING, null);
+ assertThat(actual.getStatus(), is(JobStatus.PREPARING));
+ assertThat(actual.getProgressContext().getStatus(),
is(JobStatus.PREPARING));
+ actual.setStatus(JobStatus.EXECUTE_INVENTORY_TASK);
+ assertThat(actual.getStatus(), is(JobStatus.EXECUTE_INVENTORY_TASK));
+ assertThat(actual.getProgressContext().getStatus(),
is(JobStatus.EXECUTE_INVENTORY_TASK));
+ assertThat(actual.toProgress().getStatus(),
is(JobStatus.EXECUTE_INVENTORY_TASK));
+ }
}
diff --git
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
index 9d059413078..a56e8c98a02 100644
---
a/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
+++
b/kernel/data-pipeline/scenario/migration/core/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyChecker.java
@@ -36,6 +36,7 @@ import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourc
import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
import
org.apache.shardingsphere.data.pipeline.core.exception.data.PipelineTableDataConsistencyCheckLoadingFailedException;
import
org.apache.shardingsphere.data.pipeline.core.ingest.position.type.pk.UniqueKeyIngestPosition;
+import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.core.job.id.PipelineJobIdUtils;
import
org.apache.shardingsphere.data.pipeline.core.job.progress.TransmissionJobItemProgress;
import
org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobUpdateProgress;
@@ -101,6 +102,7 @@ public final class MigrationDataConsistencyChecker
implements PipelineDataConsis
.forEach(dataNode -> sourceTableNames.add(new
QualifiedTable(dataNode.getSchemaName(), dataNode.getTableName()).format()))));
progressContext.setRecordsCount(getRecordsCount());
progressContext.getTableNames().addAll(sourceTableNames);
+ progressContext.setStatus(JobStatus.PREPARING);
progressContext.onProgressUpdated(new PipelineJobUpdateProgress(0));
Map<QualifiedTable, TableDataConsistencyCheckResult> checkResultMap =
new LinkedHashMap<>();
try (
@@ -111,6 +113,8 @@ public final class MigrationDataConsistencyChecker
implements PipelineDataConsis
if (progressContext.getTableCheckRangePositions().isEmpty()) {
progressContext.getTableCheckRangePositions().addAll(splitCrossTables());
}
+ progressContext.setStatus(JobStatus.EXECUTE_INVENTORY_TASK);
+ progressContext.onProgressUpdated(new
PipelineJobUpdateProgress(0));
for (TableCheckRangePosition each :
progressContext.getTableCheckRangePositions()) {
TableDataConsistencyCheckResult checkResult =
checkSingleTableInventoryData(each, tableChecker, dataSourceManager);
log.info("checkResult: {}, table: {}, checkRangePosition: {}",
checkResult, each.getSourceDataNode(), each);
diff --git
a/kernel/data-pipeline/scenario/migration/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/query/ShowMigrationCheckStatusExecutor.java
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/query/ShowMigrationCheckStatusExecutor.java
index 64d62110ea8..b8f46d2447b 100644
---
a/kernel/data-pipeline/scenario/migration/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/query/ShowMigrationCheckStatusExecutor.java
+++
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/query/ShowMigrationCheckStatusExecutor.java
@@ -38,8 +38,8 @@ public final class ShowMigrationCheckStatusExecutor
implements DistSQLQueryExecu
@Override
public Collection<String> getColumnNames(final
ShowMigrationCheckStatusStatement sqlStatement) {
- return Arrays.asList("tables", "result", "check_failed_tables",
"ignored_tables", "active", "inventory_finished_percentage",
"inventory_remaining_seconds", "incremental_idle_seconds",
- "check_begin_time", "check_end_time", "duration_seconds",
"algorithm_type", "algorithm_props", "error_message");
+ return Arrays.asList("tables", "result", "check_failed_tables",
"ignored_tables", "status", "active", "inventory_finished_percentage",
"inventory_remaining_seconds",
+ "incremental_idle_seconds", "check_begin_time",
"check_end_time", "duration_seconds", "algorithm_type", "algorithm_props",
"error_message");
}
@Override
@@ -50,7 +50,7 @@ public final class ShowMigrationCheckStatusExecutor
implements DistSQLQueryExecu
private LocalDataQueryResultRow convert(final ConsistencyCheckJobItemInfo
info) {
String incrementalIdleSeconds = null ==
info.getIncrementalIdleSeconds() ? "" :
String.valueOf(info.getIncrementalIdleSeconds());
- return new LocalDataQueryResultRow(info.getTableNames(),
info.getCheckSuccess(), info.getCheckFailedTableNames(),
info.getIgnoredTableNames(), info.isActive(),
+ return new LocalDataQueryResultRow(info.getTableNames(),
info.getCheckSuccess(), info.getCheckFailedTableNames(),
info.getIgnoredTableNames(), info.getStatus(), info.isActive(),
info.getInventoryFinishedPercentage(),
info.getInventoryRemainingSeconds(), incrementalIdleSeconds,
info.getCheckBeginTime(), info.getCheckEndTime(),
info.getDurationSeconds(), info.getAlgorithmType(), info.getAlgorithmProps(),
info.getErrorMessage());
}
diff --git
a/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/query/ShowMigrationCheckStatusExecutorTest.java
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/query/ShowMigrationCheckStatusExecutorTest.java
new file mode 100644
index 00000000000..f8df9502c00
--- /dev/null
+++
b/kernel/data-pipeline/scenario/migration/distsql/handler/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/distsql/handler/query/ShowMigrationCheckStatusExecutorTest.java
@@ -0,0 +1,91 @@
+/*
+ * 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.scenario.migration.distsql.handler.query;
+
+import
org.apache.shardingsphere.data.pipeline.core.consistencycheck.pojo.ConsistencyCheckJobItemInfo;
+import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
+import
org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.api.ConsistencyCheckJobAPI;
+import
org.apache.shardingsphere.data.pipeline.scenario.migration.distsql.statement.queryable.ShowMigrationCheckStatusStatement;
+import
org.apache.shardingsphere.infra.merge.result.impl.local.LocalDataQueryResultRow;
+import org.apache.shardingsphere.mode.manager.ContextManager;
+import org.junit.jupiter.api.Test;
+import org.mockito.internal.configuration.plugins.Plugins;
+
+import java.util.Arrays;
+import java.util.Collection;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+final class ShowMigrationCheckStatusExecutorTest {
+
+ private final ShowMigrationCheckStatusExecutor executor = new
ShowMigrationCheckStatusExecutor();
+
+ @Test
+ void assertGetColumnNames() {
+ Collection<String> actual = executor.getColumnNames(new
ShowMigrationCheckStatusStatement("foo_job"));
+ assertThat(actual, is(Arrays.asList("tables", "result",
"check_failed_tables", "ignored_tables", "status", "active",
"inventory_finished_percentage",
+ "inventory_remaining_seconds", "incremental_idle_seconds",
"check_begin_time", "check_end_time", "duration_seconds", "algorithm_type",
"algorithm_props", "error_message")));
+ }
+
+ @Test
+ void assertGetRows() throws ReflectiveOperationException {
+ ConsistencyCheckJobItemInfo jobItemInfo = createJobItemInfo(10L);
+ mockJobAPI(jobItemInfo);
+ Collection<LocalDataQueryResultRow> actualRows = executor.getRows(new
ShowMigrationCheckStatusStatement("foo_job"), mock(ContextManager.class));
+ assertThat(actualRows.size(), is(1));
+ assertRow(actualRows.iterator().next(), "foo_table", "true",
"foo_failed_table", "foo_ignored_table", "EXECUTE_INVENTORY_TASK", "true",
"50", "10", "10",
+ "foo_begin_time", "foo_end_time", "20", "foo_algorithm",
"foo_props", "foo_error");
+ }
+
+ private ConsistencyCheckJobItemInfo createJobItemInfo(final Long
incrementalIdleSeconds) {
+ ConsistencyCheckJobItemInfo result = new ConsistencyCheckJobItemInfo();
+ result.setTableNames("foo_table");
+ result.setCheckSuccess(true);
+ result.setCheckFailedTableNames("foo_failed_table");
+ result.setIgnoredTableNames("foo_ignored_table");
+ result.setStatus(JobStatus.EXECUTE_INVENTORY_TASK);
+ result.setActive(true);
+ result.setInventoryFinishedPercentage(50);
+ result.setInventoryRemainingSeconds(10L);
+ result.setIncrementalIdleSeconds(incrementalIdleSeconds);
+ result.setCheckBeginTime("foo_begin_time");
+ result.setCheckEndTime("foo_end_time");
+ result.setDurationSeconds(20L);
+ result.setAlgorithmType("foo_algorithm");
+ result.setAlgorithmProps("foo_props");
+ result.setErrorMessage("foo_error");
+ return result;
+ }
+
+ private void mockJobAPI(final ConsistencyCheckJobItemInfo jobItemInfo)
throws ReflectiveOperationException {
+ ConsistencyCheckJobAPI jobAPI = mock(ConsistencyCheckJobAPI.class);
+ when(jobAPI.getJobItemInfo("foo_job")).thenReturn(jobItemInfo);
+
Plugins.getMemberAccessor().set(ShowMigrationCheckStatusExecutor.class.getDeclaredField("jobAPI"),
executor, jobAPI);
+ }
+
+ private void assertRow(final LocalDataQueryResultRow actualRow, final
String... expected) {
+ for (int i = 0; i < expected.length; i++) {
+ assertThat(actualRow.getCell(i + 1), is(expected[i]));
+ }
+ assertThrows(IllegalArgumentException.class, () ->
actualRow.getCell(expected.length + 1));
+ }
+}
diff --git
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/ConsistencyCheckJobAPITest.java
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/ConsistencyCheckJobAPITest.java
index 8d5b89d3c26..496ac71bdc0 100644
---
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/ConsistencyCheckJobAPITest.java
+++
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/ConsistencyCheckJobAPITest.java
@@ -191,6 +191,7 @@ class ConsistencyCheckJobAPITest {
assertThat(actual.getCheckFailedTableNames(), is(""));
assertThat(actual.getIgnoredTableNames(), is(""));
assertThat(actual.getInventoryFinishedPercentage(), is(50));
+ assertThat(actual.getStatus(), is(JobStatus.EXECUTE_INVENTORY_TASK));
assertNotNull(actual.getCheckBeginTime());
assertNull(actual.getCheckEndTime());
}
diff --git
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyCheckerTest.java
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyCheckerTest.java
index e913f6e7e52..198ac46f85a 100644
---
a/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyCheckerTest.java
+++
b/test/it/pipeline/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/migration/check/consistency/MigrationDataConsistencyCheckerTest.java
@@ -25,6 +25,7 @@ import
org.apache.shardingsphere.data.pipeline.core.context.PipelineContextManag
import
org.apache.shardingsphere.data.pipeline.core.context.TransmissionProcessContext;
import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSource;
import
org.apache.shardingsphere.data.pipeline.core.datasource.PipelineDataSourceManager;
+import org.apache.shardingsphere.data.pipeline.core.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.core.job.api.PipelineAPIFactory;
import
org.apache.shardingsphere.data.pipeline.core.registrycenter.repository.PipelineGovernanceFacade;
import
org.apache.shardingsphere.data.pipeline.scenario.migration.config.MigrationJobConfiguration;
@@ -58,23 +59,25 @@ class MigrationDataConsistencyCheckerTest {
@Test
void assertFixtureCheck() throws SQLException {
MigrationJobConfiguration jobConfig = createJobConfiguration(true);
- Map<String, TableDataConsistencyCheckResult> checkResultMap =
check(jobConfig, "FIXTURE");
+ ConsistencyCheckJobItemProgressContext progressContext =
createConsistencyCheckJobItemProgressContext(jobConfig.getJobId());
+ Map<String, TableDataConsistencyCheckResult> checkResultMap =
check(jobConfig, "FIXTURE", progressContext);
TableDataConsistencyCheckResult actual = checkResultMap.get("t_order");
assertTrue(actual.isMatched());
assertFalse(actual.isIgnored());
+ assertThat(progressContext.getStatus(),
is(JobStatus.EXECUTE_INVENTORY_TASK));
}
@Test
void assertDataMatchCheck() throws SQLException {
MigrationJobConfiguration jobConfig = createJobConfiguration(false);
- Map<String, TableDataConsistencyCheckResult> checkResultMap =
check(jobConfig, "DATA_MATCH");
+ Map<String, TableDataConsistencyCheckResult> checkResultMap =
check(jobConfig, "DATA_MATCH",
createConsistencyCheckJobItemProgressContext(jobConfig.getJobId()));
TableDataConsistencyCheckResult actual = checkResultMap.get("t_order");
assertFalse(actual.isMatched());
assertTrue(actual.isIgnored());
assertThat(actual.getIgnoredType(),
is(TableDataConsistencyCheckIgnoredType.NO_UNIQUE_KEY));
}
- private Map<String, TableDataConsistencyCheckResult> check(final
MigrationJobConfiguration jobConfig, final String algorithmType) {
+ private Map<String, TableDataConsistencyCheckResult> check(final
MigrationJobConfiguration jobConfig, final String algorithmType, final
ConsistencyCheckJobItemProgressContext progressContext) {
JobConfigurationPOJO jobConfigurationPOJO = new JobConfigurationPOJO();
jobConfigurationPOJO.setJobParameter(YamlEngine.marshal(new
YamlMigrationJobConfigurationSwapper().swapToYamlConfiguration(jobConfig)));
jobConfigurationPOJO.setJobName(jobConfig.getJobId());
@@ -82,8 +85,7 @@ class MigrationDataConsistencyCheckerTest {
PipelineGovernanceFacade governanceFacade =
PipelineAPIFactory.getPipelineGovernanceFacade(PipelineContextUtils.getContextKey());
getClusterPersistRepository().persist(String.format("/pipeline/jobs/%s/config",
jobConfig.getJobId()), YamlEngine.marshal(jobConfigurationPOJO));
governanceFacade.getJobItemFacade().getProcess().persist(jobConfig.getJobId(),
0, "");
- return new MigrationDataConsistencyChecker(jobConfig, new
TransmissionProcessContext(jobConfig.getJobId(), null),
-
createConsistencyCheckJobItemProgressContext(jobConfig.getJobId())).check(algorithmType,
null);
+ return new MigrationDataConsistencyChecker(jobConfig, new
TransmissionProcessContext(jobConfig.getJobId(), null),
progressContext).check(algorithmType, null);
}
private ClusterPersistRepository getClusterPersistRepository() {
