You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "loserwang1024 (via GitHub)" <gi...@apache.org> on 2025/03/18 01:57:12 UTC
[PR] [FLINK-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
loserwang1024 opened a new pull request, #3954:
URL: https://github.com/apache/flink-cdc/pull/3954
Fix [FLINK-37484](https://issues.apache.org/jira/browse/FLINK-37484), unevenly chunk split also enables unbounded-chunk-first.enabled.
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#discussion_r2024012227
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/ChunkRange.java:
##########
@@ -83,4 +83,9 @@ public boolean equals(Object o) {
public int hashCode() {
return Objects.hash(chunkStart, chunkEnd);
}
+
+ @Override
+ public String toString() {
+ return "ChunkRange{" + "chunkStart=" + chunkStart + ", chunkEnd=" + chunkEnd + '}';
Review Comment:
```suggestion
return "ChunkRange{chunkStart=" + chunkStart + ", chunkEnd=" + chunkEnd + '}';
```
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#issuecomment-2771411199
> @loserwang1024 Do we need to update the docs about this option?
It seems that docs have already included this option., this pr just improve it.
```
<td>scan.incremental.snapshot.unbounded-chunk-first.enabled</td>
<td>optional</td>
<td style="word-wrap: break-word;">false</td>
<td>Boolean</td>
<td>
Whether to assign the unbounded chunks first during snapshot reading phase.<br>
This might help reduce the risk of the TaskManager experiencing an out-of-memory (OOM) error when taking a snapshot of the largest unbounded chunk.<br>
Experimental option, defaults to false.
</td>
```
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on code in PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#discussion_r2005063000
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java:
##########
@@ -366,16 +366,20 @@ private void splitTable(TableId nextTable) {
hasRecordSchema = true;
tableSchemas.putAll(splits.iterator().next().getTableSchemas());
}
- final List<SchemalessSnapshotSplit> schemalessSnapshotSplits =
- splits.stream()
- .map(SnapshotSplit::toSchemalessSnapshotSplit)
- .collect(Collectors.toList());
+ List<String> splitIds = new ArrayList<>();
+ for (SnapshotSplit split : splits) {
+ SchemalessSnapshotSplit schemalessSnapshotSplit =
+ split.toSchemalessSnapshotSplit();
+ splitIds.add(schemalessSnapshotSplit.splitId());
+ if (sourceConfig.isAssignUnboundedChunkFirst() && split.getSplitEnd() == null) {
Review Comment:
This is not effective for MongoDB. Now MongoDB still advances the unbounded chunk when splitting chunks. Could it need to be optimized?https://github.com/apache/flink-cdc/blob/5edd199d5fad7ba09bb54644a54e7efb2d177454/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SplitVectorSplitStrategy.java#L134-L138
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#issuecomment-2771245720
<img width="1060" alt="image" src="https://github.com/user-attachments/assets/f16b62a7-d46f-4d8d-8675-668df7dd677a" />
test fail with timeout. This PR doesn't modify mongo cdc, please help rerun test again.
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#discussion_r2006919004
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java:
##########
@@ -366,16 +366,20 @@ private void splitTable(TableId nextTable) {
hasRecordSchema = true;
tableSchemas.putAll(splits.iterator().next().getTableSchemas());
}
- final List<SchemalessSnapshotSplit> schemalessSnapshotSplits =
- splits.stream()
- .map(SnapshotSplit::toSchemalessSnapshotSplit)
- .collect(Collectors.toList());
+ List<String> splitIds = new ArrayList<>();
+ for (SnapshotSplit split : splits) {
+ SchemalessSnapshotSplit schemalessSnapshotSplit =
+ split.toSchemalessSnapshotSplit();
+ splitIds.add(schemalessSnapshotSplit.splitId());
+ if (sourceConfig.isAssignUnboundedChunkFirst() && split.getSplitEnd() == null) {
Review Comment:
It's hard to add a test from mongo cdc which chunk size is MB rather than number of record. The test which you have added also just returns one snapshot split.It can not prove unbounded-chunk-first.enabled.
Any I noticed that currently mongo cdc is event and un-evently share same logical, no need to modify it. And this pr will not influence mongo cdc
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on code in PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#discussion_r2013441153
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java:
##########
@@ -202,7 +202,20 @@ public void testReadSingleTableWithSingleParallelismAndSkipBackFill() throws Exc
FailoverPhase.NEVER,
new String[] {tableName},
RestartStrategies.fixedDelayRestart(1, 0),
- true);
+ Collections.singletonMap("scan.incremental.snapshot.backfill.skip", "true"));
+ }
+
+ @Test
+ public void testReadSingleTableAndUnboundedChunkFirst() throws Exception {
+ testMySqlParallelSource(
+ 1,
+ DEFAULT_SCAN_STARTUP_MODE,
+ FailoverType.NONE,
+ FailoverPhase.NEVER,
+ new String[] {tableName},
+ RestartStrategies.fixedDelayRestart(1, 0),
+ Collections.singletonMap(
+ "scan.incremental.snapshot.unbounded-chunk-first.enabled", "true"));
Review Comment:
`scan.incremental.snapshot.unbounded-chunk-first.enabled` is already set by `assignEndingFirst == null`.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/ChunkRange.java:
##########
@@ -83,4 +83,9 @@ public boolean equals(Object o) {
public int hashCode() {
return Objects.hash(chunkStart, chunkEnd);
}
+
+ @Override
+ public String toString() {
+ return "ChunkRange{" + "chunkStart=" + chunkStart + ", chunkEnd=" + chunkEnd + '}';
Review Comment:
```suggestion
return "ChunkRange{chunkStart=" + chunkStart + ", chunkEnd=" + chunkEnd + '}';
```
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "beryllw (via GitHub)" <gi...@apache.org>.
beryllw commented on PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#issuecomment-2739563589
Looks like the CI failure [here](https://github.com/apache/flink-cdc/actions/runs/13914366724/job/38936122888?pr=3954) relates to [org.apache.flink.cdc.connectors.mysql.source.assigners.MySqlChunkSplitterTest#testSplitEvenlySizedChunksEndingFirst](https://github.com/apache/flink-cdc/blob/5edd199d5fad7ba09bb54644a54e7efb2d177454/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitterTest.java#L91-L117). Could we adjust or delete this test to fix 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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#issuecomment-2753656272
Hi @loserwang1024, could you please rebase this PR with latest `master`?
Code style verifier has been updated to enforce JUnit 5 + AssertJ framework and these classes might need to be migrated:
* JUnit 4 style test annotations should be changed to JUnit 5 equivalents
* `org.junit.Test` => `org.junit.jupiter.api.Test`
* `@Before`, `@BeforeClass` => `@BeforeEach`, `@BeforeAll`
* `@After`, `@AfterClass` => `@AfterEach`, `@AfterAll`
* JUnit Assertions / Hamcrest Assertions are not allowed, including:
* `org.junit.Assert`
* `org.junit.jupiter.api.Assertions`
* `org.hamcrest.*`
Use `org.assertj.core.api.Assertions` instead.
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 merged PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#issuecomment-2731481011
@lvyanquan @leonardBang , CC
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "ruanhang1993 (via GitHub)" <gi...@apache.org>.
ruanhang1993 commented on PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#issuecomment-2771253621
@loserwang1024 Do we need to update the docs about this option?
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#discussion_r2022252129
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlSourceITCase.java:
##########
@@ -202,7 +202,20 @@ public void testReadSingleTableWithSingleParallelismAndSkipBackFill() throws Exc
FailoverPhase.NEVER,
new String[] {tableName},
RestartStrategies.fixedDelayRestart(1, 0),
- true);
+ Collections.singletonMap("scan.incremental.snapshot.backfill.skip", "true"));
+ }
+
+ @Test
+ public void testReadSingleTableAndUnboundedChunkFirst() throws Exception {
+ testMySqlParallelSource(
+ 1,
+ DEFAULT_SCAN_STARTUP_MODE,
+ FailoverType.NONE,
+ FailoverPhase.NEVER,
+ new String[] {tableName},
+ RestartStrategies.fixedDelayRestart(1, 0),
+ Collections.singletonMap(
+ "scan.incremental.snapshot.unbounded-chunk-first.enabled", "true"));
Review Comment:
ok, it seems that new test is no need to add again.
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#issuecomment-2768355162
Rebase this pr to master, @ruanhang1993 , CC
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#discussion_r2006730420
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java:
##########
@@ -366,16 +366,20 @@ private void splitTable(TableId nextTable) {
hasRecordSchema = true;
tableSchemas.putAll(splits.iterator().next().getTableSchemas());
}
- final List<SchemalessSnapshotSplit> schemalessSnapshotSplits =
- splits.stream()
- .map(SnapshotSplit::toSchemalessSnapshotSplit)
- .collect(Collectors.toList());
+ List<String> splitIds = new ArrayList<>();
+ for (SnapshotSplit split : splits) {
+ SchemalessSnapshotSplit schemalessSnapshotSplit =
+ split.toSchemalessSnapshotSplit();
+ splitIds.add(schemalessSnapshotSplit.splitId());
+ if (sourceConfig.isAssignUnboundedChunkFirst() && split.getSplitEnd() == null) {
Review Comment:
@beryllw thanks a lot. I was busy last several days, and will fix it today
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#discussion_r2006919004
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java:
##########
@@ -366,16 +366,20 @@ private void splitTable(TableId nextTable) {
hasRecordSchema = true;
tableSchemas.putAll(splits.iterator().next().getTableSchemas());
}
- final List<SchemalessSnapshotSplit> schemalessSnapshotSplits =
- splits.stream()
- .map(SnapshotSplit::toSchemalessSnapshotSplit)
- .collect(Collectors.toList());
+ List<String> splitIds = new ArrayList<>();
+ for (SnapshotSplit split : splits) {
+ SchemalessSnapshotSplit schemalessSnapshotSplit =
+ split.toSchemalessSnapshotSplit();
+ splitIds.add(schemalessSnapshotSplit.splitId());
+ if (sourceConfig.isAssignUnboundedChunkFirst() && split.getSplitEnd() == null) {
Review Comment:
It's hard to add a test from mongo cdc which chunk size is MB rather than number of record. The test which you have added also just returns one snapshot split.It can not prove unbounded-chunk-first.enabled
--
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-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #3954:
URL: https://github.com/apache/flink-cdc/pull/3954#discussion_r2006919004
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java:
##########
@@ -366,16 +366,20 @@ private void splitTable(TableId nextTable) {
hasRecordSchema = true;
tableSchemas.putAll(splits.iterator().next().getTableSchemas());
}
- final List<SchemalessSnapshotSplit> schemalessSnapshotSplits =
- splits.stream()
- .map(SnapshotSplit::toSchemalessSnapshotSplit)
- .collect(Collectors.toList());
+ List<String> splitIds = new ArrayList<>();
+ for (SnapshotSplit split : splits) {
+ SchemalessSnapshotSplit schemalessSnapshotSplit =
+ split.toSchemalessSnapshotSplit();
+ splitIds.add(schemalessSnapshotSplit.splitId());
+ if (sourceConfig.isAssignUnboundedChunkFirst() && split.getSplitEnd() == null) {
Review Comment:
It's hard to add a test from mongo cdc which chunk size is MB rather than number of record. The test which you have added also just returns one snapshot split.It can not prove unbounded-chunk-first.enabled.
Any I noticed that currently mongo cdc is event and un-evently share same logical, no need to modify 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