You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by az...@apache.org on 2023/08/17 08:27:11 UTC
[shardingsphere] branch master updated: Isolate consistency check tableCheckPositions for source and target (#28143)
This is an automated email from the ASF dual-hosted git repository.
azexin 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 ab5d911478a Isolate consistency check tableCheckPositions for source and target (#28143)
ab5d911478a is described below
commit ab5d911478a3d6f91e94f237efbab5fb0b9bdac0
Author: Hongsheng Zhong <zh...@apache.org>
AuthorDate: Thu Aug 17 16:27:05 2023 +0800
Isolate consistency check tableCheckPositions for source and target (#28143)
* Refactor tableCheckPositions to sourceTableCheckPositions and targetTableCheckPositions
* Add ConsistencyCheckJobItemContextTest
---
.../progress/ConsistencyCheckJobItemProgress.java | 6 +-
.../yaml/YamlConsistencyCheckJobItemProgress.java | 5 +-
...YamlConsistencyCheckJobItemProgressSwapper.java | 13 ++---
.../ConsistencyCheckJobItemProgressContext.java | 4 +-
.../table/MatchingTableDataConsistencyChecker.java | 10 ++--
.../api/impl/ConsistencyCheckJobAPI.java | 3 +-
.../context/ConsistencyCheckJobItemContext.java | 3 +-
.../ConsistencyCheckJobItemContextTest.java | 65 ++++++++++++++++++++++
.../consistencycheck/ConsistencyCheckJobTest.java | 6 +-
9 files changed, 92 insertions(+), 23 deletions(-)
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/ConsistencyCheckJobItemProgress.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/ConsistencyCheckJobItemProgress.java
index 04950d53919..3b36f63e16c 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/ConsistencyCheckJobItemProgress.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/ConsistencyCheckJobItemProgress.java
@@ -28,10 +28,10 @@ import java.util.Map;
/**
* Data consistency check job item progress.
*/
-// TODO move package
@RequiredArgsConstructor
@Getter
@ToString
+// TODO Refactor structure, List<TableProgress>
public final class ConsistencyCheckJobItemProgress implements PipelineJobItemProgress {
@Setter
@@ -49,5 +49,7 @@ public final class ConsistencyCheckJobItemProgress implements PipelineJobItemPro
private final Long checkEndTimeMillis;
- private final Map<String, Object> tableCheckPositions;
+ private final Map<String, Object> sourceTableCheckPositions;
+
+ private final Map<String, Object> targetTableCheckPositions;
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgress.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgress.java
index 3e5b9086484..c7984ca0c4e 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgress.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgress.java
@@ -21,6 +21,7 @@ import lombok.Getter;
import lombok.Setter;
import org.apache.shardingsphere.infra.util.yaml.YamlConfiguration;
+import java.util.LinkedHashMap;
import java.util.Map;
/**
@@ -44,5 +45,7 @@ public final class YamlConsistencyCheckJobItemProgress implements YamlConfigurat
private Long checkEndTimeMillis;
- private Map<String, Object> tableCheckPositions;
+ private Map<String, Object> sourceTableCheckPositions = new LinkedHashMap<>();
+
+ private Map<String, Object> targetTableCheckPositions = new LinkedHashMap<>();
}
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgressSwapper.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgressSwapper.java
index 96a28986c9d..bece1b9b17c 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgressSwapper.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/common/job/progress/yaml/YamlConsistencyCheckJobItemProgressSwapper.java
@@ -21,9 +21,6 @@ import org.apache.shardingsphere.data.pipeline.common.job.JobStatus;
import org.apache.shardingsphere.data.pipeline.common.job.progress.ConsistencyCheckJobItemProgress;
import org.apache.shardingsphere.infra.util.yaml.swapper.YamlConfigurationSwapper;
-import java.util.LinkedHashMap;
-import java.util.Map;
-
/**
* YAML data check job item progress swapper.
*/
@@ -39,18 +36,16 @@ public final class YamlConsistencyCheckJobItemProgressSwapper implements YamlCon
result.setRecordsCount(data.getRecordsCount());
result.setCheckBeginTimeMillis(data.getCheckBeginTimeMillis());
result.setCheckEndTimeMillis(data.getCheckEndTimeMillis());
- result.setTableCheckPositions(data.getTableCheckPositions());
+ result.setSourceTableCheckPositions(data.getSourceTableCheckPositions());
+ result.setTargetTableCheckPositions(data.getTargetTableCheckPositions());
return result;
}
@Override
public ConsistencyCheckJobItemProgress swapToObject(final YamlConsistencyCheckJobItemProgress yamlConfig) {
- Map<String, Object> tableCheckPositions = new LinkedHashMap<>();
- if (null != yamlConfig.getTableCheckPositions()) {
- tableCheckPositions.putAll(yamlConfig.getTableCheckPositions());
- }
ConsistencyCheckJobItemProgress result = new ConsistencyCheckJobItemProgress(yamlConfig.getTableNames(), yamlConfig.getIgnoredTableNames(), yamlConfig.getCheckedRecordsCount(),
- yamlConfig.getRecordsCount(), yamlConfig.getCheckBeginTimeMillis(), yamlConfig.getCheckEndTimeMillis(), tableCheckPositions);
+ yamlConfig.getRecordsCount(), yamlConfig.getCheckBeginTimeMillis(), yamlConfig.getCheckEndTimeMillis(),
+ yamlConfig.getSourceTableCheckPositions(), yamlConfig.getTargetTableCheckPositions());
result.setStatus(JobStatus.valueOf(yamlConfig.getStatus()));
return result;
}
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 f11b5d9c8d4..7822408a81c 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
@@ -54,7 +54,9 @@ public final class ConsistencyCheckJobItemProgressContext implements PipelineJob
private volatile Long checkEndTimeMillis;
- private final Map<String, Object> tableCheckPositions = new ConcurrentHashMap<>();
+ private final Map<String, Object> sourceTableCheckPositions = new ConcurrentHashMap<>();
+
+ private final Map<String, Object> targetTableCheckPositions = new ConcurrentHashMap<>();
@Override
public void onProgressUpdated(final PipelineJobProgressUpdatedParameter param) {
diff --git a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableDataConsistencyChecker.java b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableDataConsistencyChecker.java
index b88f7bf2d17..79518e3edb4 100644
--- a/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableDataConsistencyChecker.java
+++ b/kernel/data-pipeline/core/src/main/java/org/apache/shardingsphere/data/pipeline/core/consistencycheck/table/MatchingTableDataConsistencyChecker.java
@@ -36,7 +36,6 @@ import org.apache.shardingsphere.infra.util.close.QuietlyCloser;
import java.sql.SQLException;
import java.util.Iterator;
-import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutionException;
@@ -65,11 +64,10 @@ public abstract class MatchingTableDataConsistencyChecker implements TableDataCo
}
private TableDataConsistencyCheckResult checkSingleTableInventoryData(final TableDataConsistencyCheckParameter param, final ThreadPoolExecutor executor) {
- Map<String, Object> tableCheckPositions = param.getProgressContext().getTableCheckPositions();
SingleTableInventoryCalculateParameter sourceParam = new SingleTableInventoryCalculateParameter(param.getSourceDataSource(), param.getSourceTable(),
- param.getColumnNames(), param.getUniqueKeys(), tableCheckPositions.get(param.getSourceTable().getTableName().getOriginal()));
+ param.getColumnNames(), param.getUniqueKeys(), param.getProgressContext().getSourceTableCheckPositions().get(param.getSourceTable().getTableName().getOriginal()));
SingleTableInventoryCalculateParameter targetParam = new SingleTableInventoryCalculateParameter(param.getTargetDataSource(), param.getTargetTable(),
- param.getColumnNames(), param.getUniqueKeys(), tableCheckPositions.get(param.getTargetTable().getTableName().getOriginal()));
+ param.getColumnNames(), param.getUniqueKeys(), param.getProgressContext().getTargetTableCheckPositions().get(param.getTargetTable().getTableName().getOriginal()));
SingleTableInventoryCalculator calculator = getSingleTableInventoryCalculator();
Iterator<SingleTableInventoryCalculatedResult> sourceCalculatedResults = waitFuture(executor.submit(() -> calculator.calculate(sourceParam))).iterator();
Iterator<SingleTableInventoryCalculatedResult> targetCalculatedResults = waitFuture(executor.submit(() -> calculator.calculate(targetParam))).iterator();
@@ -99,10 +97,10 @@ public abstract class MatchingTableDataConsistencyChecker implements TableDataCo
break;
}
if (sourceCalculatedResult.getMaxUniqueKeyValue().isPresent()) {
- param.getProgressContext().getTableCheckPositions().put(param.getSourceTable().getTableName().getOriginal(), sourceCalculatedResult.getMaxUniqueKeyValue().get());
+ param.getProgressContext().getSourceTableCheckPositions().put(param.getSourceTable().getTableName().getOriginal(), sourceCalculatedResult.getMaxUniqueKeyValue().get());
}
if (targetCalculatedResult.getMaxUniqueKeyValue().isPresent()) {
- param.getProgressContext().getTableCheckPositions().put(param.getTargetTable().getTableName().getOriginal(), targetCalculatedResult.getMaxUniqueKeyValue().get());
+ param.getProgressContext().getTargetTableCheckPositions().put(param.getTargetTable().getTableName().getOriginal(), targetCalculatedResult.getMaxUniqueKeyValue().get());
}
param.getProgressContext().onProgressUpdated(new PipelineJobProgressUpdatedParameter(sourceCalculatedResult.getRecordsCount()));
}
diff --git a/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java b/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java
index 361d045d6df..6473566fe4f 100644
--- a/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java
+++ b/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/api/impl/ConsistencyCheckJobAPI.java
@@ -149,7 +149,8 @@ public final class ConsistencyCheckJobAPI extends AbstractPipelineJobAPIImpl {
String tableNames = String.join(",", progressContext.getTableNames());
String ignoredTableNames = String.join(",", progressContext.getIgnoredTableNames());
ConsistencyCheckJobItemProgress jobItemProgress = new ConsistencyCheckJobItemProgress(tableNames, ignoredTableNames, progressContext.getCheckedRecordsCount().get(),
- progressContext.getRecordsCount(), progressContext.getCheckBeginTimeMillis(), progressContext.getCheckEndTimeMillis(), progressContext.getTableCheckPositions());
+ progressContext.getRecordsCount(), progressContext.getCheckBeginTimeMillis(), progressContext.getCheckEndTimeMillis(),
+ progressContext.getSourceTableCheckPositions(), progressContext.getTargetTableCheckPositions());
jobItemProgress.setStatus(context.getStatus());
return YamlEngine.marshal(swapper.swapToYamlConfiguration(jobItemProgress));
}
diff --git a/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContext.java b/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContext.java
index fa2bce7cdba..329d9d1e8cb 100644
--- a/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContext.java
+++ b/kernel/data-pipeline/scenario/consistencycheck/src/main/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContext.java
@@ -59,7 +59,8 @@ public final class ConsistencyCheckJobItemContext implements PipelineJobItemCont
progressContext = new ConsistencyCheckJobItemProgressContext(jobId, shardingItem);
if (null != jobItemProgress) {
progressContext.getCheckedRecordsCount().set(Optional.ofNullable(jobItemProgress.getCheckedRecordsCount()).orElse(0L));
- Optional.ofNullable(jobItemProgress.getTableCheckPositions()).ifPresent(progressContext.getTableCheckPositions()::putAll);
+ Optional.ofNullable(jobItemProgress.getSourceTableCheckPositions()).ifPresent(progressContext.getSourceTableCheckPositions()::putAll);
+ Optional.ofNullable(jobItemProgress.getTargetTableCheckPositions()).ifPresent(progressContext.getTargetTableCheckPositions()::putAll);
}
processContext = new ConsistencyCheckProcessContext(jobId);
}
diff --git a/kernel/data-pipeline/scenario/consistencycheck/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContextTest.java b/kernel/data-pipeline/scenario/consistencycheck/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContextTest.java
new file mode 100644
index 00000000000..32f4ac70e18
--- /dev/null
+++ b/kernel/data-pipeline/scenario/consistencycheck/src/test/java/org/apache/shardingsphere/data/pipeline/scenario/consistencycheck/context/ConsistencyCheckJobItemContextTest.java
@@ -0,0 +1,65 @@
+/*
+ * 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.consistencycheck.context;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.shardingsphere.data.pipeline.common.job.JobStatus;
+import org.apache.shardingsphere.data.pipeline.common.job.progress.ConsistencyCheckJobItemProgress;
+import org.apache.shardingsphere.data.pipeline.core.consistencycheck.ConsistencyCheckJobItemProgressContext;
+import org.apache.shardingsphere.data.pipeline.scenario.consistencycheck.config.ConsistencyCheckJobConfiguration;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+
+class ConsistencyCheckJobItemContextTest {
+
+ private static final String TABLE = "t_order";
+
+ @Test
+ void assertConstructWithoutTableCheckPositions() {
+ Map<String, Object> sourceTableCheckPositions = Collections.emptyMap();
+ Map<String, Object> targetTableCheckPositions = Collections.emptyMap();
+ ConsistencyCheckJobItemProgress jobItemProgress = new ConsistencyCheckJobItemProgress(TABLE, null, 0L, 10L, null, null, sourceTableCheckPositions, targetTableCheckPositions);
+ ConsistencyCheckJobItemContext actual = new ConsistencyCheckJobItemContext(new ConsistencyCheckJobConfiguration("", "", "DATA_MATCH", null), 0, JobStatus.RUNNING, jobItemProgress);
+ verifyProgressContext(actual.getProgressContext(), 0, sourceTableCheckPositions, targetTableCheckPositions);
+ }
+
+ @Test
+ void assertConstructWithTableCheckPositions() {
+ Map<String, Object> sourceTableCheckPositions = ImmutableMap.of(TABLE, 6);
+ Map<String, Object> targetTableCheckPositions = ImmutableMap.of(TABLE, 5);
+ ConsistencyCheckJobItemProgress jobItemProgress = new ConsistencyCheckJobItemProgress(TABLE, null, 0L, 10L, null, null, sourceTableCheckPositions, targetTableCheckPositions);
+ ConsistencyCheckJobItemContext actual = new ConsistencyCheckJobItemContext(new ConsistencyCheckJobConfiguration("", "", "DATA_MATCH", null), 0, JobStatus.RUNNING, jobItemProgress);
+ verifyProgressContext(actual.getProgressContext(), 1, sourceTableCheckPositions, targetTableCheckPositions);
+ assertThat(actual.getProgressContext().getSourceTableCheckPositions().get(TABLE), is(6));
+ assertThat(actual.getProgressContext().getTargetTableCheckPositions().get(TABLE), is(5));
+ }
+
+ private void verifyProgressContext(final ConsistencyCheckJobItemProgressContext progressContext, final int expectedSize,
+ final Map<String, Object> sourceTableCheckPositions, final Map<String, Object> targetTableCheckPositions) {
+ assertThat(progressContext.getSourceTableCheckPositions().size(), is(expectedSize));
+ assertThat(progressContext.getTargetTableCheckPositions().size(), is(expectedSize));
+ assertNotSame(progressContext.getSourceTableCheckPositions(), sourceTableCheckPositions);
+ assertNotSame(progressContext.getTargetTableCheckPositions(), targetTableCheckPositions);
+ }
+}
diff --git a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobTest.java b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobTest.java
index c4e90a89381..613a74c9bbf 100644
--- a/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobTest.java
+++ b/test/it/pipeline/src/test/java/org/apache/shardingsphere/test/it/data/pipeline/scenario/consistencycheck/ConsistencyCheckJobTest.java
@@ -56,13 +56,15 @@ class ConsistencyCheckJobTest {
ConsistencyCheckJob consistencyCheckJob = new ConsistencyCheckJob(checkJobId);
ConsistencyCheckJobItemContext actual = consistencyCheckJob.buildPipelineJobItemContext(
new ShardingContext(checkJobId, "", 1, YamlEngine.marshal(createYamlConsistencyCheckJobConfiguration(checkJobId)), 0, ""));
- assertThat(actual.getProgressContext().getTableCheckPositions(), is(expectTableCheckPosition));
+ assertThat(actual.getProgressContext().getSourceTableCheckPositions(), is(expectTableCheckPosition));
+ assertThat(actual.getProgressContext().getTargetTableCheckPositions(), is(expectTableCheckPosition));
}
private YamlConsistencyCheckJobItemProgress createYamlConsistencyCheckJobItemProgress(final Map<String, Object> expectTableCheckPosition) {
YamlConsistencyCheckJobItemProgress result = new YamlConsistencyCheckJobItemProgress();
result.setStatus(JobStatus.RUNNING.name());
- result.setTableCheckPositions(expectTableCheckPosition);
+ result.setSourceTableCheckPositions(expectTableCheckPosition);
+ result.setTargetTableCheckPositions(expectTableCheckPosition);
return result;
}