1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.shardingsphere.data.pipeline.core.consistencycheck;
19
20 import lombok.Getter;
21 import lombok.RequiredArgsConstructor;
22 import lombok.Setter;
23 import org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobProgressListener;
24 import org.apache.shardingsphere.data.pipeline.core.job.progress.listener.PipelineJobProgressUpdatedParameter;
25 import org.apache.shardingsphere.data.pipeline.core.job.progress.persist.PipelineJobProgressPersistService;
26
27 import java.util.Collection;
28 import java.util.Map;
29 import java.util.concurrent.ConcurrentHashMap;
30 import java.util.concurrent.CopyOnWriteArraySet;
31 import java.util.concurrent.atomic.AtomicLong;
32
33
34
35
36 @RequiredArgsConstructor
37 @Getter
38 @Setter
39 public final class ConsistencyCheckJobItemProgressContext implements PipelineJobProgressListener {
40
41 private final String jobId;
42
43 private final int shardingItem;
44
45 private final Collection<String> tableNames = new CopyOnWriteArraySet<>();
46
47 private final Collection<String> ignoredTableNames = new CopyOnWriteArraySet<>();
48
49 private volatile long recordsCount;
50
51 private final AtomicLong checkedRecordsCount = new AtomicLong(0L);
52
53 private final long checkBeginTimeMillis = System.currentTimeMillis();
54
55 private volatile Long checkEndTimeMillis;
56
57 private final Map<String, Object> sourceTableCheckPositions = new ConcurrentHashMap<>();
58
59 private final Map<String, Object> targetTableCheckPositions = new ConcurrentHashMap<>();
60
61 private final String sourceDatabaseType;
62
63 @Override
64 public void onProgressUpdated(final PipelineJobProgressUpdatedParameter param) {
65 checkedRecordsCount.addAndGet(param.getProcessedRecordsCount());
66 PipelineJobProgressPersistService.notifyPersist(jobId, shardingItem);
67 }
68 }