[FLINK-37484][cdc-connector] unevenly chunk split also enables unbounded-chunk-first.enabled. (#3954)

Author: loserwang1024

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) {
+                        // assign unbounded split first
+                        remainingSplits.add(0, schemalessSnapshotSplit);
+                    } else {
+                        remainingSplits.add(schemalessSnapshotSplit);
+                    }
+                }
+
                 chunkNum += splits.size();
-                remainingSplits.addAll(schemalessSnapshotSplits);
-                List<String> splitIds =
-                        schemalessSnapshotSplits.stream()
-                                .map(SchemalessSnapshotSplit::splitId)
-                                .collect(Collectors.toList());
                 enumeratorMetrics.getTableMetrics(nextTable).addNewSplits(splitIds);
 
                 if (!chunkSplitter.hasNextChunk()) {

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 + '}';
+    }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java

@@ -469,15 +469,8 @@ private List<ChunkRange> splitEvenlySizedChunks(
                 break;
             }
         }
-        // add the unbounded split
-        // assign unbounded split first, both the largest and smallest unbounded chunks are
-        // completed
-        // in the first two splits
-        if (sourceConfig.isAssignUnboundedChunkFirst()) {
-            splits.add(0, ChunkRange.of(chunkStart, null));
-        } else {
-            splits.add(ChunkRange.of(chunkStart, null));
-        }
+        // add the ending split
+        splits.add(ChunkRange.of(chunkStart, null));
         return splits;
     }
 

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 + '}';
+    }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitter.java

@@ -316,15 +316,9 @@ public List<ChunkRange> splitEvenlySizedChunks(
                 break;
             }
         }
-        // add the unbounded split
-        // assign unbounded split first, both the largest and smallest unbounded chunks are
-        // completed
-        // in the first two splits
-        if (sourceConfig.isAssignUnboundedChunkFirst()) {
-            splits.add(0, ChunkRange.of(chunkStart, null));
-        } else {
-            splits.add(ChunkRange.of(chunkStart, null));
-        }
+
+        // add the ending split
+        splits.add(ChunkRange.of(chunkStart, null));
         return splits;
     }
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

@@ -324,12 +324,19 @@ private void splitTable(TableId nextTable) {
                     tableSchema.putAll(splits.iterator().next().getTableSchemas());
                     tableSchemas.putAll(tableSchema);
                 }
-                final List<MySqlSchemalessSnapshotSplit> schemaLessSnapshotSplits =
-                        splits.stream()
-                                .map(MySqlSnapshotSplit::toSchemalessSnapshotSplit)
-                                .collect(Collectors.toList());
+
+                for (MySqlSnapshotSplit split : splits) {
+                    MySqlSchemalessSnapshotSplit schemalessSnapshotSplit =
+                            split.toSchemalessSnapshotSplit();
+                    if (sourceConfig.isAssignUnboundedChunkFirst() && split.getSplitEnd() == null) {
+                        // assign unbounded split first
+                        remainingSplits.add(0, schemalessSnapshotSplit);
+                    } else {
+                        remainingSplits.add(schemalessSnapshotSplit);
+                    }
+                }
+
                 chunkNum += splits.size();
-                remainingSplits.addAll(schemaLessSnapshotSplits);
                 if (!chunkSplitter.hasNextChunk()) {
                     remainingTables.remove(nextTable);
                 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlChunkSplitterTest.java

@@ -85,34 +85,4 @@ void testSplitEvenlySizedChunksNormal() {
                         ChunkRange.of(2147483637, 2147483647),
                         ChunkRange.of(2147483647, null));
     }
-
-    @Test
-    public void testSplitEvenlySizedChunksEndingFirst() {
-        MySqlSourceConfig sourceConfig =
-                new MySqlSourceConfigFactory()
-                        .startupOptions(StartupOptions.initial())
-                        .databaseList("")
-                        .tableList("")
-                        .hostname("")
-                        .username("")
-                        .password("")
-                        .serverTimeZone(ZoneId.of("UTC").toString())
-                        .assignUnboundedChunkFirst(true)
-                        .createConfig(0);
-        MySqlChunkSplitter splitter = new MySqlChunkSplitter(null, sourceConfig);
-
-        List<ChunkRange> res =
-                splitter.splitEvenlySizedChunks(
-                        new TableId("catalog", "db", "tab"),
-                        Integer.MAX_VALUE - 20,
-                        Integer.MAX_VALUE,
-                        20,
-                        10,
-                        10);
-        Assertions.assertThat(res)
-                .containsExactly(
-                        ChunkRange.of(2147483647, null),
-                        ChunkRange.of(null, 2147483637),
-                        ChunkRange.of(2147483637, 2147483647));
-    }
 }