Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[mysql] Fix newly added table invalid request meta group id error(#1327) #2482

Merged
merged 4 commits into from
Sep 14, 2023
Merged

[mysql] Fix newly added table invalid request meta group id error(#1327) #2482

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
0bf00b4
[mysql] Fix newly added table invalid request meta group id error(#1327)
qg-lin Sep 12, 2023
38ced58
[mysql] Fix newly added table invalid request meta group id error(#1327)
qg-lin Sep 13, 2023
50758dc
[mysql] add comment for PR(#2482)
qg-lin Sep 14, 2023
d8479a5
[hotfix][mysql] Keep assigned splits in order to fix wrong meta group…
leonardBang Sep 14, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 16 additions & 7 deletions .../java/com/ververica/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,13 +43,14 @@

import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
Expand Down Expand Up @@ -106,7 +107,7 @@ public MySqlSnapshotSplitAssigner(
currentParallelism,
new ArrayList<>(),
new ArrayList<>(),
new HashMap<>(),
new LinkedHashMap<>(),
new HashMap<>(),
new HashMap<>(),
AssignerStatus.INITIAL_ASSIGNING,
Expand Down Expand Up @@ -152,7 +153,17 @@ private MySqlSnapshotSplitAssigner(
this.currentParallelism = currentParallelism;
this.alreadyProcessedTables = alreadyProcessedTables;
this.remainingSplits = new CopyOnWriteArrayList<>(remainingSplits);
this.assignedSplits = assignedSplits;
// When job restore from savepoint, sort the existing tables and newly added tables
// to let enumerator only send newly added tables' BinlogSplitMetaEvent
this.assignedSplits =
assignedSplits.entrySet().stream()
.sorted(Entry.comparingByKey())
.collect(
Collectors.toMap(
Entry::getKey,
Entry::getValue,
(o, o2) -> o,
LinkedHashMap::new));
ruanhang1993 marked this conversation as resolved.
Show resolved Hide resolved
this.tableSchemas = tableSchemas;
this.splitFinishedOffsets = splitFinishedOffsets;
this.assignerStatus = assignerStatus;
Expand Down Expand Up @@ -232,7 +243,7 @@ private void captureNewlyAddedTables() {

// remove unassigned tables/splits if it does not satisfy new table filter
List<String> splitsToRemove = new LinkedList<>();
for (Map.Entry<String, MySqlSchemalessSnapshotSplit> splitEntry :
for (Entry<String, MySqlSchemalessSnapshotSplit> splitEntry :
assignedSplits.entrySet()) {
if (tablesToRemove.contains(splitEntry.getValue().getTableId())) {
splitsToRemove.add(splitEntry.getKey());
Expand Down Expand Up @@ -367,9 +378,7 @@ public List<FinishedSnapshotSplitInfo> getFinishedSplitInfos() {
"The assigner is not ready to offer finished split information, this should not be called");
}
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
assignedSplits.values().stream()
.sorted(Comparator.comparing(MySqlSplit::splitId))
.collect(Collectors.toList());
new ArrayList<>(assignedSplits.values());
List<FinishedSnapshotSplitInfo> finishedSnapshotSplitInfos = new ArrayList<>();
for (MySqlSchemalessSnapshotSplit split : assignedSnapshotSplit) {
BinlogOffset binlogOffset = splitFinishedOffsets.get(split.splitId());
Expand Down
16 changes: 10 additions & 6 deletions ...cdc/src/main/java/com/ververica/cdc/connectors/mysql/source/reader/MySqlSourceReader.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -467,20 +467,24 @@ private MySqlBinlogSplit discoverTableSchemasForBinlogSplit(

private Set<String> getExistedSplitsOfLastGroup(
List<FinishedSnapshotSplitInfo> finishedSnapshotSplits, int metaGroupSize) {
Set<String> existedSplitsOfLastGroup = new HashSet<>();
int splitsNumOfLastGroup =
finishedSnapshotSplits.size() % sourceConfig.getSplitMetaGroupSize();
if (splitsNumOfLastGroup != 0) {
int lastGroupStart =
((int) (finishedSnapshotSplits.size() / sourceConfig.getSplitMetaGroupSize()))
* metaGroupSize;
existedSplitsOfLastGroup =
finishedSnapshotSplits
.subList(lastGroupStart, lastGroupStart + splitsNumOfLastGroup).stream()
// Keep same order with MySqlHybridSplitAssigner.createBinlogSplit() to avoid
// 'invalid request meta group id' error
List<String> sortedFinishedSnapshotSplits =
finishedSnapshotSplits.stream()
.map(FinishedSnapshotSplitInfo::getSplitId)
.collect(Collectors.toSet());
.sorted()
.collect(Collectors.toList());
return new HashSet<>(
sortedFinishedSnapshotSplits.subList(
lastGroupStart, lastGroupStart + splitsNumOfLastGroup));
}
return existedSplitsOfLastGroup;
return new HashSet<>();
}

private void logCurrentBinlogOffsets(List<MySqlSplit> splits, long checkpointId) {
Expand Down
Loading