[postgres][hotfix] Fix pass wrong schemaChangeEnabled.

Author: loserwang1024

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-postgres/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresPipelineRecordEmitter.java

@@ -135,7 +135,7 @@ protected void processElement(
             maybeSendCreateTableEventFromCache(tableId, output);
         } else if (isDataChangeRecord(element)) {
             handleDataChangeRecord(element, output);
-        } else if (isSchemaChangeEvent(element) && sourceConfig.isSchemaChangeEnabled()) {
+        } else if (isSchemaChangeEvent(element) && sourceConfig.isIncludeSchemaChanges()) {
             handleSchemaChangeRecord(element, output, splitState);
         }
         super.processElement(element, output, splitState);

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

@@ -620,7 +620,7 @@ public void testSchemaChangeWithDataInserts() throws Exception {
         configFactory.database(inventoryDatabase.getDatabaseName());
         configFactory.slotName(slotName);
         configFactory.decodingPluginName("pgoutput");
-        configFactory.enableSchemaChange(true);
+        configFactory.includeSchemaChanges(true);
 
         FlinkSourceProvider sourceProvider =
                 (FlinkSourceProvider)

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfig.java

@@ -40,7 +40,6 @@ public class PostgresSourceConfig extends JdbcSourceConfig {
     private final int lsnCommitCheckpointsDelay;
     private final boolean includePartitionedTables;
     private final boolean includeDatabaseInTableId;
-    private final boolean schemaChangeEnabled;
 
     public PostgresSourceConfig(
             int subtaskId,
@@ -72,8 +71,7 @@ public PostgresSourceConfig(
             int lsnCommitCheckpointsDelay,
             boolean assignUnboundedChunkFirst,
             boolean includePartitionedTables,
-            boolean includeDatabaseInTableId,
-            boolean schemaChangeEnabled) {
+            boolean includeDatabaseInTableId) {
         super(
                 startupOptions,
                 databaseList,
@@ -105,7 +103,6 @@ public PostgresSourceConfig(
         this.lsnCommitCheckpointsDelay = lsnCommitCheckpointsDelay;
         this.includePartitionedTables = includePartitionedTables;
         this.includeDatabaseInTableId = includeDatabaseInTableId;
-        this.schemaChangeEnabled = schemaChangeEnabled;
     }
 
     /**
@@ -159,9 +156,4 @@ public PostgresConnectorConfig getDbzConnectorConfig() {
     public boolean isIncludeDatabaseInTableId() {
         return includeDatabaseInTableId;
     }
-
-    /** Returns whether to infer column types via JDBC TypeRegistry on schema change events. */
-    public boolean isSchemaChangeEnabled() {
-        return schemaChangeEnabled;
-    }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceConfigFactory.java

@@ -57,8 +57,6 @@ public class PostgresSourceConfigFactory extends JdbcSourceConfigFactory {
     private boolean includeDatabaseInTableId =
             PostgresSourceOptions.TABLE_ID_INCLUDE_DATABASE.defaultValue();
 
-    private boolean schemaChangeEnabled = false;
-
     /** Creates a new {@link PostgresSourceConfig} for the given subtask {@code subtaskId}. */
     @Override
     public PostgresSourceConfig create(int subtaskId) {
@@ -142,8 +140,7 @@ public PostgresSourceConfig create(int subtaskId) {
                 lsnCommitCheckpointsDelay,
                 assignUnboundedChunkFirst,
                 includePartitionedTables,
-                includeDatabaseInTableId,
-                schemaChangeEnabled);
+                includeDatabaseInTableId);
     }
 
     /**
@@ -201,9 +198,4 @@ public void setIncludePartitionedTables(boolean includePartitionedTables) {
     public void setIncludeDatabaseInTableId(boolean includeDatabaseInTableId) {
         this.includeDatabaseInTableId = includeDatabaseInTableId;
     }
-
-    /** Set whether to infer schema change event on relation message. */
-    public void enableSchemaChange(boolean schemaChangeEnabled) {
-        this.schemaChangeEnabled = schemaChangeEnabled;
-    }
 }

flink-cdc-e2e-tests/flink-cdc-pipeline-e2e-tests/src/test/java/org/apache/flink/cdc/pipeline/tests/PostgresE2eITCase.java

@@ -116,12 +116,14 @@ void testSyncWholeDatabase() throws Exception {
                                 + "  scan.startup.mode: initial\n"
                                 + "  server-time-zone: UTC\n"
                                 + "  connect.timeout: 120s\n"
+                                + "  schema-change.enabled: true\n"
                                 + "\n"
                                 + "sink:\n"
                                 + "  type: values\n"
                                 + "\n"
                                 + "pipeline:\n"
-                                + "  parallelism: %d",
+                                + "  parallelism: %d\n"
+                                + "  schema.change.behavior: evolve",
                         INTER_CONTAINER_POSTGRES_ALIAS,
                         5432,
                         POSTGRES_TEST_USER,
@@ -172,5 +174,47 @@ void testSyncWholeDatabase() throws Exception {
             LOG.error("Update table for CDC failed.", e);
             throw new RuntimeException(e);
         }
+
+        LOG.info("Begin schema change stage.");
+
+        try (Connection conn =
+                        getJdbcConnection(
+                                POSTGRES_CONTAINER, postgresInventoryDatabase.getDatabaseName());
+                Statement stat = conn.createStatement()) {
+            // Test ADD COLUMN
+            stat.execute("ALTER TABLE inventory.products ADD COLUMN category VARCHAR(255);");
+            stat.execute(
+                    "INSERT INTO inventory.products VALUES (default, 'widget', 'A small widget', 1.5, 'tools');");
+
+            // Test DROP COLUMN
+            stat.execute("ALTER TABLE inventory.products DROP COLUMN weight;");
+            stat.execute(
+                    "INSERT INTO inventory.products VALUES (default, 'gadget', 'A useful gadget', 'electronics');");
+
+            // Test RENAME COLUMN
+            stat.execute(
+                    "ALTER TABLE inventory.products RENAME COLUMN category TO product_category;");
+            stat.execute(
+                    "INSERT INTO inventory.products VALUES (default, 'gizmo', 'A fancy gizmo', 'gadgets');");
+        } catch (Exception e) {
+            LOG.error("Schema change test failed.", e);
+            throw new RuntimeException(e);
+        }
+
+        // Validate schema change events and corresponding data
+        waitUntilSpecificEvent(
+                "AddColumnEvent{tableId=inventory.products, addedColumns=[ColumnWithPosition{column=`category` VARCHAR(255), position=LAST, existedColumnName=null}]}");
+        waitUntilSpecificEvent(
+                "DataChangeEvent{tableId=inventory.products, before=[], after=[110, widget, A small widget, 1.5, tools], op=INSERT, meta=()}");
+
+        waitUntilSpecificEvent(
+                "DropColumnEvent{tableId=inventory.products, droppedColumnNames=[weight]}");
+        waitUntilSpecificEvent(
+                "DataChangeEvent{tableId=inventory.products, before=[], after=[111, gadget, A useful gadget, electronics], op=INSERT, meta=()}");
+
+        waitUntilSpecificEvent(
+                "RenameColumnEvent{tableId=inventory.products, nameMapping={category=product_category}}");
+        waitUntilSpecificEvent(
+                "DataChangeEvent{tableId=inventory.products, before=[], after=[112, gizmo, A fancy gizmo, gadgets], op=INSERT, meta=()}");
     }
 }