[FLINK-38840][postgres] Spotless fixes

Author: bakjos

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresSourceFetchTaskContext.java

@@ -127,37 +127,41 @@ public void configure(SourceSplitBase sourceSplitBase) {
         PostgresConnectorConfig dbzConfig = getDbzConnectorConfig();
         if (sourceSplitBase instanceof SnapshotSplit) {
 
-            Configuration.Builder builder = dbzConfig
-                    .getConfig()
-                    .edit()
-                    .with(
-                            "table.include.list",
-                            getTableList(
-                                    ((SnapshotSplit) sourceSplitBase).getTableId()))
-                    .with(
-                            SLOT_NAME.name(),
-                            ((PostgresSourceConfig) sourceConfig)
-                                    .getSlotNameForBackfillTask())
-                    // drop slot for backfill stream split
-                    .with(DROP_SLOT_ON_STOP.name(), true)
-                    // Disable heartbeat event in snapshot split fetcher
-                    .with(Heartbeat.HEARTBEAT_INTERVAL, 0);
+            Configuration.Builder builder =
+                    dbzConfig
+                            .getConfig()
+                            .edit()
+                            .with(
+                                    "table.include.list",
+                                    getTableList(((SnapshotSplit) sourceSplitBase).getTableId()))
+                            .with(
+                                    SLOT_NAME.name(),
+                                    ((PostgresSourceConfig) sourceConfig)
+                                            .getSlotNameForBackfillTask())
+                            // drop slot for backfill stream split
+                            .with(DROP_SLOT_ON_STOP.name(), true)
+                            // Disable heartbeat event in snapshot split fetcher
+                            .with(Heartbeat.HEARTBEAT_INTERVAL, 0);
 
             try {
-                String autoCreateMode = dbzConfig.getConfig().getString(PUBLICATION_AUTOCREATE_MODE.name());
+                String autoCreateMode =
+                        dbzConfig.getConfig().getString(PUBLICATION_AUTOCREATE_MODE.name());
                 if ("filtered".equalsIgnoreCase(autoCreateMode)) {
-                    // disable publication auto create for snapshot split when auto create mode is filtered
+                    // disable publication auto create for snapshot split when auto create mode is
+                    // filtered
                     // Prevent backfill slots from modifying the shared publication.
-                    // Without this, each backfill slot would SET the publication to only include its current table
-                    builder = builder.with(PUBLICATION_AUTOCREATE_MODE.name(),
-                            PostgresConnectorConfig.AutoCreateMode.DISABLED);
+                    // Without this, each backfill slot would SET the publication to only include
+                    // its current table
+                    builder =
+                            builder.with(
+                                    PUBLICATION_AUTOCREATE_MODE.name(),
+                                    PostgresConnectorConfig.AutoCreateMode.DISABLED);
                 }
             } catch (Exception e) {
                 // ignore
             }
 
-            dbzConfig =
-                    new PostgresConnectorConfig(builder.build());
+            dbzConfig = new PostgresConnectorConfig(builder.build());
         } else {
 
             Configuration.Builder builder = dbzConfig.getConfig().edit();

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/PostgresSourceITCase.java

@@ -1053,8 +1053,10 @@ private void checkStreamDataForFilteredPublication(
             collectedSnapshotData.add(rowStr);
 
             // Insert test record right after Table 3 starts its snapshot phase
-            // This ensures the insert happens while Table 3's backfill slot may have altered the publication
-            if (collectedSnapshotData.size() == recordsWhenTable3Starts && !insertedDuringSnapshot) {
+            // This ensures the insert happens while Table 3's backfill slot may have altered the
+            // publication
+            if (collectedSnapshotData.size() == recordsWhenTable3Starts
+                    && !insertedDuringSnapshot) {
                 Thread.sleep(1000L); // Give Table 3's backfill slot time to alter the publication
                 try (PostgresConnection connection = getConnection()) {
                     connection.setAutoCommit(false);
@@ -1090,9 +1092,11 @@ private void checkStreamDataForFilteredPublication(
             }
         }
 
-        boolean insertFound = collectedStreamData.contains("+I[9999, test_user, TestCity, 123456789]");
+        boolean insertFound =
+                collectedStreamData.contains("+I[9999, test_user, TestCity, 123456789]");
         assertThat(insertFound)
-                .as("The insert into Customers table during Table 3 snapshot phase should be captured.")
+                .as(
+                        "The insert into Customers table during Table 3 snapshot phase should be captured.")
                 .isTrue();
     }