You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "morozov (via GitHub)" <gi...@apache.org> on 2025/11/03 06:14:18 UTC
[PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
morozov opened a new pull request, #4087:
URL: https://github.com/apache/flink-cdc/pull/4087
The root cause is that the binlog split metadata transfer protocol relies on the order of finished snapshot split infos to be stable and corresponding to the order of split assignment (the infos of newly added/snapshotted tables are appended to the end of the list). However, when `MySqlSnapshotSplitAssigner` is restored from state, `assignedSplits` are reordered, which breaks this assumption.
## Change summary
1. **Require assigned snapshot splits to be ordered**. This isn't strictly necessary to fix the bug but follows directly from the JavaDoc I added to `MySqlSnapshotSplitAssigner#assignedSplits`. If the order is important, the type should guarantee that it's preserved. Note the changes in the deserialization code. Not using an ordered map there while the order is important may cause other hard to diagnose issues.
2. **Rely on stable order of assigned splits**. Instead of identifying duplicate received split infos by split ID, ignore the first N elements that we know we already have.
3. **Eliminate code duplication in MySqlBinlogSplit constructors**. There are currently two constructors where one doesn't call the other. The subsequent commit adds a check that needs to be enforced regardless of which of the constructors was used, so I'm combining them.
4. **Enforce no duplicate finished snapshot split infos in `MySqlBinlogSplit`**. By design, a binlog split cannot contain duplicate finished snapshot split infos. If it does, it indicates the fact that it was constructed incorrectly. If it happens, it's a bug, and we want to fail as early as possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-4095710471
https://github.com/apache/flink-cdc/blob/c4b698bffb692f480622123fec36494ec505929c/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReader.java#L291
If I'm reading the code correctly, if we cannot find the corresponding `matchedSplit`, it's `null`, so `BinlogSplitReader#shouldEmit()` will return `false`, which leads to data loss, not duplication. But this is a different issue ([FLINK-38270](https://issues.apache.org/jira/browse/FLINK-38270)).
The principal differences are:
1. Here, the connection gets stuck not producing any data; there, it transitions to reading the binlog but drops data from tables.
2. This one is reproducible by addition of a table; the other one requires addition and removal during snapshot.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2951261588
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
Assuming the shards saved by the Enumerator are `split3`, `split2`, `split1`, they will be sorted in the original way as `split1`, `split2`, `split3`.
If the Source Reader of a job has already obtained `split3` and `split2` according to the original sorting method, and the job has been upgraded in version, requesting the third shard according to the latest implementation method will request `split3` again. Will there be compatibility issues with this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-3635483559
@leonardBang could you restart the tests? The logs are no longer available.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2960034889
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
> ### Before Upgrade (Old Version)
>
> 3. **Reader has already received partial splits** based on the sorted order, e.g., `[split1, split2]`
This doesn't look right. The reader has received the splits in the same order as they were on the enumerator before the checkpoint, unsorted, so it will be `[split3, split2]`.
> ### After Upgrade (New Version)
>
> 5. **Reader still has state `[split1, split2]`** (unchanged from checkpoint)
The same. How is this possible? The reader doesn't sort the elements.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov closed pull request #4087: [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission
URL: https://github.com/apache/flink-cdc/pull/4087
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-3761317048
@leonardBang, @lvyanquan do you still want to get this issue fixed? I closed the PR due to the lack of interest but you reopened it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-3169381982
I'm not sure how to test this. The issue is reproducible if a source is restarted mid-snapshot of a newly added table and requires consuming the changes in the new table from the binlog. Could maintainers recommend an existing test on top of which I could build this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2954320584
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
> Assuming the shards saved by the Enumerator are `split3`, `split2`, `split1`, they will be sorted in the original way as `split1`, `split2`, `split3`.
> If the Source Reader of a job has already obtained `split1` and `split2` \[...]
Is this possible?
1. Finished snapshot split infos are built in the order of split assignment: https://github.com/apache/flink-cdc/blob/2e82317810f1fe62bc75d4bc22247668d9613f20/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java#L397-L399
2. They are sent to the source reader in the same order: https://github.com/apache/flink-cdc/blob/4bab36707b781d2221d4d5c6b515b074b72789ec/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java#L304-L306
This is true for both the pre-upgrade and post-upgrade versions.
So if before restoring the job from a checkpoint, the enumerator has the splits/infos in the order of `split3`, `split2`, `split1`, the source reader will have a slice of this list (e.g. `split3`, `split2`) but not `split1`, `split2`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2909892178
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/split/MySqlBinlogSplit.java:
##########
@@ -58,6 +62,19 @@ public MySqlBinlogSplit(
int totalFinishedSplitSize,
boolean isSuspended) {
super(splitId);
+
+ Set<String> seenSplitIds = new HashSet<>();
+ for (FinishedSnapshotSplitInfo splitInfo : finishedSnapshotSplitInfos) {
+ if (seenSplitIds.contains(splitInfo.getSplitId())) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Found duplicate split ID %s in finished snapshot split infos",
+ splitInfo.getSplitId()));
+ }
+
+ seenSplitIds.add(splitInfo.getSplitId());
+ }
Review Comment:
Please extract this part of the code into a separate method to encapsulate it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-3637858687
@leonardBang, after the rebase, the build is green.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan merged PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2958034741
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
Thank you @morozov for the detailed explanation. However, I believe there's still a compatibility issue that wasn't fully addressed. Let me clarify my concern with a concrete **job upgrade scenario**.
## The Scenario
Consider an **existing running job** that is being upgraded from the old version to the new version:
### Before Upgrade (Old Version)
1. Splits are assigned in order: `split3 → split2 → split1`
2. **Enumerator restores from checkpoint and sorts by split ID:**
```java
// Old code sorts on restore
.sorted(Entry.comparingByKey())
```
Result: `assignedSplits = [split1, split2, split3]`
3. **Reader has already received partial splits** based on the sorted order, e.g., `[split1, split2]`
4. `finishedSnapshotSplitInfos` on Enumerator side is built from sorted order: `[split1, split2, split3]`
5. If `chunk-meta.group.size = 2`, the partitions are:
```
[
0: [split1, split2]
1: [split3]
]
```
6. Reader has received group 0, next request will be for group 1 (`split3`)
### After Upgrade (New Version)
1. **Job restarts from checkpoint and upgrades to new version**
2. **Enumerator restores WITHOUT sorting:**
```java
// New code does NOT sort on restore
this.assignedSplits = assignedSplits;
```
Result: `assignedSplits = [split3, split2, split1]` (preserves original insertion order)
3. `finishedSnapshotSplitInfos` is now built from unsorted order: `[split3, split2, split1]`
4. The partitions are now:
```
[
0: [split3, split2]
1: [split1]
]
```
5. **Reader still has state `[split1, split2]`** (unchanged from checkpoint)
6. Reader requests group 1 (because `receivedMetaNum = 2`, and `2 / 2 = 1`)
7. Enumerator returns `[split1]` instead of the expected `split3`
8. Reader calculates `expectedNumberOfAlreadyRetrievedElements = 2 % 2 = 0`
9. Reader discards 0 elements and appends `[split1]` to `[split1, split2]`
10. **Result:** Reader state becomes `[split1, split2, split1]` — **contains duplicate `split1` and missing `split3`**
## The Root Cause
The issue is that the **order semantics changed** between versions, but the **state serializer version was NOT incremented**:
```java
private static final int VERSION = 5; // Still 5, not incremented
```
The old version and new version interpret the same serialized state differently:
- Old version: sorts on restore → order is by split ID
- New version: does not sort on restore → order is by insertion time
This breaks the invariant that both Enumerator and Reader must use the same order.
## Suggested Fix
1. **Increment the state serializer version:**
```java
private static final int VERSION = 6;
```
2. **Add migration logic for version 5 and below:**
```java
private SnapshotPendingSplitsState deserializeSnapshotPendingSplitsState(
int version, int splitVersion, DataInputDeserializer in) throws IOException {
// ... existing deserialization logic ...
// For old versions, sort the assigned splits to maintain compatibility
if (version < 6) {
LinkedHashMap<String, MySqlSchemalessSnapshotSplit> sortedSplits =
assignedSchemalessSnapshotSplits.entrySet().stream()
.sorted(Map.Entry.comparingByKey())
.collect(Collectors.toMap(
Map.Entry::getKey,
Map.Entry::getValue,
(o1, o2) -> o1,
LinkedHashMap::new));
assignedSchemalessSnapshotSplits = sortedSplits;
}
// ...
}
```
This ensures backward compatibility for existing jobs upgrading to the new version.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2957250835
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
Sorry, there was an error in the previous description. It was due to the inconsistency in the order before and after, and we only have a fixed sequence before. Will this result in duplicate requests for splits?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2954320584
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
> Assuming the shards saved by the Enumerator are `split3`, `split2`, `split1`, they will be sorted in the original way as `split1`, `split2`, `split3`.
> If the Source Reader of a job has already obtained `split1` and `split2` \[...]
Is this possible?
1. Finished snapshot split infos are built in the order of split assignment: https://github.com/apache/flink-cdc/blob/2e82317810f1fe62bc75d4bc22247668d9613f20/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java#L397-L399
2. They are sent to the source reader in the same order: https://github.com/apache/flink-cdc/blob/4bab36707b781d2221d4d5c6b515b074b72789ec/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java#L304-L306
This is true for both the pre-upgrade and post-upgrade versions.
So if before restoring the job from a checkpoint, the enumerator has the splits/infos in the order of `split3`, `split2`, `split1`, the source reader will have a slice of this list (e.g. `split3`, `split2`) but not `split1` and `split2`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-4095003662
Hi @morozov.
I understand the problem you described now. I'd like to confirm that similar issues can also occur in scenarios other than adding new tables. Another possible scenario is that due to missing some table splits, when we can't find the corresponding matchedSplit in BinlogSplitReader#shouldEmit and proceed to emit the data anyway, it leads to data duplication.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-4095986874
Hi @ruanhang1993, Because this is a change on the critical path, you can also continue to review it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "lvyanquan (via GitHub)" <gi...@apache.org>.
lvyanquan commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-3822842497
Hi @morozov, apologies for the delayed response. As this change impacts a critical code path, we are proceeding with utmost caution. I will review and verify both the issue and the proposed fix this week.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#issuecomment-3635503905
Hey, @morozov Looks like Azure cannot re-trigger expired CI, could you rebase your PR to latest master branch to trigger new CI ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2954320584
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
> Assuming the shards saved by the Enumerator are `split3`, `split2`, `split1`, they will be sorted in the original way as `split1`, `split2`, `split3`.
> If the Source Reader of a job has already obtained `split1` and `split2` \[...]
Is this possible?
1. Finished snapshot split infos are built in the order of split assignment: https://github.com/apache/flink-cdc/blob/2e82317810f1fe62bc75d4bc22247668d9613f20/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java#L397-L399
2. They are sent to the source reader in the same order: https://github.com/apache/flink-cdc/blob/4bab36707b781d2221d4d5c6b515b074b72789ec/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java#L304-L306
This is true for both the pre-upgrade and post-upgrade versions.
So if before restoring the job from a checkpoint, the enumerator has the splits/infos in the order of `split3`, `split2`, `split1`, the source reader will have a prefix of this list (e.g. `split3`, `split2`) but not `split1` and `split2`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2960034889
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
> ### Before Upgrade (Old Version)
>
> 3\. **Reader has already received partial splits** based on the sorted order, e.g., `[split1, split2]`
This doesn't look right. The reader has received the splits in the same order as they were on the enumerator before the checkpoint, unsorted, so it will be `[split3, split2]`.
> ### After Upgrade (New Version)
>
> 5. **Reader still has state `[split1, split2]`** (unchanged from checkpoint)
The same. How is this possible? The reader doesn't sort the elements.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-38218] Fix MySQL CDC binlog split metadata split transmission [flink-cdc]
Posted by "morozov (via GitHub)" <gi...@apache.org>.
morozov commented on code in PR #4087:
URL: https://github.com/apache/flink-cdc/pull/4087#discussion_r2957366409
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java:
##########
@@ -208,9 +206,7 @@ public void close() {
private MySqlBinlogSplit createBinlogSplit() {
final List<MySqlSchemalessSnapshotSplit> assignedSnapshotSplit =
- snapshotSplitAssigner.getAssignedSplits().values().stream()
- .sorted(Comparator.comparing(MySqlSplit::splitId))
- .collect(Collectors.toList());
+ new ArrayList<>(snapshotSplitAssigner.getAssignedSplits().values());
Review Comment:
> requesting the third shard according to the latest implementation method will request `split3` again. Will there be compatibility issues with this?
It will not request `split3` again. The reader will know that there are 3 split infos in total and it has 2, so it needs to request the remaining one.
**Let's assume that `chunk-meta.group.size = 2`.**
1. The source reader state is `[split3, split2]` (`receivedMetaNum` = 2)
2. The next meta group ID is 1 (`1 / 1 = 1`): https://github.com/apache/flink-cdc/blob/441eec81a1629ee101edd3ed3ab38bcefd65db9a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/utils/ChunkUtils.java#L110-L113
3. The splits on the enumerator are partitioned by 2: https://github.com/apache/flink-cdc/blob/4bab36707b781d2221d4d5c6b515b074b72789ec/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java#L304-L306
```js
[
0: [split3, split2]
1: [split1]
]
```
4. The source reader requests group 1, the enumerator returns `[split1]`
5. The expected number of already retrieved elements in the last group on the reader is 0 (`2 % 2 = 0`): https://github.com/apache/flink-cdc/blob/2e82317810f1fe62bc75d4bc22247668d9613f20/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReader.java#L424-L426
6. The source reader discards 0 first elements from the response, so `[split1]` remains `[split1]`.
7. The source reader appends `[split1]` to `[split3, split2]`, so the resulting source reader state becomes `[split3, split2, split1]` – correct.
**Let's assume that `chunk-meta.group.size = 3`.**
1. The source reader state is `[split3, split2]` (`receivedMetaNum` = 2)
2. The next meta group ID is 0 (`2 / 3 = 0`)
3. The splits on the enumerator are partitioned by 3:
```js
[
0: [split3, split2, split1]
]
```
4. The source reader requests group 0, the enumerator returns `[split3, split2, split1]`
5. The expected number of already retrieved elements in the last group on the reader is 2 (`2 % 3 = 2`):
6. The source reader discards 2 first elements from the response, so `[split3, split2, split1]` becomes `[split1]`.
7. The source reader appends `[split1]` to `[split3, split2]`, so the resulting source reader state becomes `[split3, split2, split1]` – correct.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org