fixed checkstyle

Author: Mrart

docs/content/docs/connectors/flink-sources/mysql-cdc.md

@@ -874,6 +874,30 @@ There are two places that need to be taken care of.
   * If no update operation is performed on the specified column, the exactly-once semantics is ensured.
   * If the update operation is performed on the specified column, only the at-least-once semantics is ensured. However, you can specify primary keys at downstream and perform the idempotence operation to ensure data correctness.
 
+#### Warning
+
+Using a **non-primary key column** as the `scan.incremental.snapshot.chunk.key-column` for a MySQL table with primary keys may lead to data inconsistencies. Below is a scenario illustrating this issue and recommendations to mitigate potential problems.
+
+#### Problem Scenario
+
+- **Table Structure:**
+    - **Primary Key:** `id`
+    - **Chunk Key Column:** `pid` (Not a primary key)
+
+- **Snapshot Splits:**
+    - **Split 0:** `1 < pid <= 3`
+    - **Split 1:** `3 < pid <= 5`
+
+- **Operation:**
+    - Two different subtasks are reading Split 0 and Split 1 concurrently.
+    - An update operation changes `pid` from `2` to `4` for `id=0` while both splits are being read. This update occurs between the low and high watermark of both splits.
+
+- **Result:**
+    - **Split 0:** Contains the record `[id=0, pid=2]`
+    - **Split 1:** Contains the record `[id=0, pid=4]`
+
+Since the order of processing these records cannot be guaranteed, the final value of `pid` for `id=0` may end up being either `2` or `4`, leading to potential data inconsistencies.
+
 ### About converting binary type data to base64 encoded data
 
 ```sql

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/debezium/reader/BinlogSplitReaderTest.java

@@ -570,7 +570,7 @@ void testReadBinlogFromEarliestOffset() throws Exception {
                     "+I[1017, user_18, Shanghai, 123567891234]",
                     "+I[1018, user_19, Shanghai, 123567891234]",
                     "+I[1019, user_20, Shanghai, 123567891234]",
-                    "+I[2000, user_21, Shanghai, 123567891234]",
+                    "+I[2000, user_21, China:Shanghai, 123567891234]",
                     "-U[103, user_3, Shanghai, 123567891234]",
                     "+U[103, user_3, Hangzhou, 123567891234]",
                     "-D[102, user_2, Shanghai, 123567891234]",

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

@@ -106,7 +106,6 @@
 import static org.apache.flink.api.common.JobStatus.RUNNING;
 import static org.apache.flink.util.Preconditions.checkState;
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.jupiter.api.Assertions.assertTrue;
 
 /** IT tests for {@link MySqlSource}. */
 @Timeout(value = 300, unit = TimeUnit.SECONDS)
@@ -777,7 +776,8 @@ public void testSnapshotFilters(String tableName, String chunkColumnName) throws
         assertEqualsInAnyOrder(
                 Arrays.asList(expectedSnapshotData),
                 fetchRowData(iterator, expectedSnapshotData.length));
-        assertTrue(!hasNextData(iterator));
+        assertThat(hasNextData(iterator)).isFalse();
+
         jobClient.cancel().get();
     }
 
@@ -855,7 +855,7 @@ public void testSnapshotFiltersEscape(String tableName, String chunkColumnName)
         assertEqualsInAnyOrder(
                 Arrays.asList(expectedSnapshotData),
                 fetchRowData(iterator, expectedSnapshotData.length));
-        assertTrue(!hasNextData(iterator));
+        assertThat(hasNextData(iterator)).isFalse();
         jobClient.cancel().get();
     }
 

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/reader/MySqlSourceReaderTest.java

@@ -184,7 +184,7 @@ void testRemoveTableUsingStateFromSnapshotPhase() throws Exception {
                     "+I[1016, user_17, Shanghai, 123567891234]",
                     "+I[1019, user_20, Shanghai, 123567891234]",
                     "+I[1018, user_19, Shanghai, 123567891234]",
-                    "+I[2000, user_21, Shanghai, 123567891234]"
+                    "+I[2000, user_21, China:Shanghai, 123567891234]"
                 };
         // Step 2: wait the snapshot splits finished reading
         List<String> actualRecords = consumeSnapshotRecords(reader, dataType);
@@ -295,7 +295,7 @@ void testFinishedUnackedSplitsUsingStateFromSnapshotPhase(boolean skipBackFill)
                     "+I[1016, user_17, Shanghai, 123567891234]",
                     "+I[1019, user_20, Shanghai, 123567891234]",
                     "+I[1018, user_19, Shanghai, 123567891234]",
-                    "+I[2000, user_21, Shanghai, 123567891234]"
+                    "+I[2000, user_21, China:Shanghai, 123567891234]"
                 };
         // Step 2: wait the snapshot splits finished reading
         List<String> actualRecords = consumeSnapshotRecords(reader, dataType);

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/utils/SnapshotFilterUtilsTest.java

@@ -18,13 +18,12 @@
 package org.apache.flink.cdc.connectors.mysql.source.utils;
 
 import io.debezium.relational.TableId;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
 
 import java.util.HashMap;
 import java.util.Map;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
+import static org.assertj.core.api.Assertions.assertThat;
 
 /** Unit test for {@link org.apache.flink.cdc.connectors.mysql.source.utils.SnapshotFilterUtils}. */
 public class SnapshotFilterUtilsTest {
@@ -34,14 +33,16 @@ public void test() {
         Map<String, String> map = new HashMap<>();
         map.put("db.user", "id > 100");
         map.put("db.order_[0-9]+", "id > 200");
-        assertEquals(
-                "id > 100", SnapshotFilterUtils.getSnapshotFilter(map, TableId.parse("db.user")));
-        assertEquals(
-                "id > 200",
-                SnapshotFilterUtils.getSnapshotFilter(map, TableId.parse("db.order_1")));
-        assertEquals(
-                "id > 200",
-                SnapshotFilterUtils.getSnapshotFilter(map, TableId.parse("db.order_2")));
-        assertNull(SnapshotFilterUtils.getSnapshotFilter(map, TableId.parse("db.shop")));
+
+        assertThat(SnapshotFilterUtils.getSnapshotFilter(map, TableId.parse("db.user")))
+                .isEqualTo("id > 100");
+
+        assertThat(SnapshotFilterUtils.getSnapshotFilter(map, TableId.parse("db.order_1")))
+                .isEqualTo("id > 200");
+
+        assertThat(SnapshotFilterUtils.getSnapshotFilter(map, TableId.parse("db.order_2")))
+                .isEqualTo("id > 200");
+
+        assertThat(SnapshotFilterUtils.getSnapshotFilter(map, TableId.parse("db.shop"))).isNull();
     }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/table/MySqlTableSourceFactoryTest.java

@@ -58,7 +58,7 @@
 import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST;
 import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.SCAN_SNAPSHOT_FETCH_SIZE;
 import static org.apache.flink.cdc.connectors.mysql.source.config.MySqlSourceOptions.USE_LEGACY_JSON_FORMAT;
-import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
 
 /** Test for {@link MySqlTableSource} created by {@link MySqlTableSourceFactory}. */
 class MySqlTableSourceFactoryTest {
@@ -485,7 +485,7 @@ void testStartupFromEarliestOffset() {
                         SCAN_INCREMENTAL_SNAPSHOT_UNBOUNDED_CHUNK_FIRST.defaultValue(),
                         false,
                         null);
-        assertEquals(expectedSource, actualSource);
+        assertThat(actualSource).isEqualTo(expectedSource);
     }
 
     @Test