[minor][common] Improve the changelogMode expression (#1793)

apache · Dec 5, 2022 · 8f920ac · 8f920ac
1 parent 8bf6160
commit 8f920ac
Show file tree

Hide file tree

Showing 5 changed files with 5 additions and 35 deletions.
diff --git a/...onnector-db2-cdc/src/main/java/com/ververica/cdc/connectors/db2/table/Db2TableSource.java b/...onnector-db2-cdc/src/main/java/com/ververica/cdc/connectors/db2/table/Db2TableSource.java
@@ -26,7 +26,6 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.types.RowKind;
 
 import com.ververica.cdc.connectors.db2.Db2Source;
 import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
@@ -93,12 +92,7 @@ public Db2TableSource(
 
     @Override
     public ChangelogMode getChangelogMode() {
-        return ChangelogMode.newBuilder()
-                .addContainedKind(RowKind.INSERT)
-                .addContainedKind(RowKind.DELETE)
-                .addContainedKind(RowKind.UPDATE_AFTER)
-                .addContainedKind(RowKind.UPDATE_BEFORE)
-                .build();
+        return ChangelogMode.all();
     }
 
     @Override

diff --git a/...or-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/table/MySqlTableSource.java b/...or-mysql-cdc/src/main/java/com/ververica/cdc/connectors/mysql/table/MySqlTableSource.java
@@ -27,7 +27,6 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.types.RowKind;
 
 import com.ververica.cdc.connectors.mysql.source.MySqlSource;
 import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
@@ -147,12 +146,7 @@ public MySqlTableSource(
 
     @Override
     public ChangelogMode getChangelogMode() {
-        return ChangelogMode.newBuilder()
-                .addContainedKind(RowKind.INSERT)
-                .addContainedKind(RowKind.UPDATE_BEFORE)
-                .addContainedKind(RowKind.UPDATE_AFTER)
-                .addContainedKind(RowKind.DELETE)
-                .build();
+        return ChangelogMode.all();
     }
 
     @Override

diff --git a/...-cdc/src/main/java/com/ververica/cdc/connectors/oceanbase/table/OceanBaseTableSource.java b/...-cdc/src/main/java/com/ververica/cdc/connectors/oceanbase/table/OceanBaseTableSource.java
@@ -26,7 +26,6 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.types.RowKind;
 
 import com.ververica.cdc.connectors.oceanbase.OceanBaseSource;
 import com.ververica.cdc.connectors.oceanbase.source.RowDataOceanBaseDeserializationSchema;
@@ -124,12 +123,7 @@ public OceanBaseTableSource(
 
     @Override
     public ChangelogMode getChangelogMode() {
-        return ChangelogMode.newBuilder()
-                .addContainedKind(RowKind.INSERT)
-                .addContainedKind(RowKind.UPDATE_BEFORE)
-                .addContainedKind(RowKind.UPDATE_AFTER)
-                .addContainedKind(RowKind.DELETE)
-                .build();
+        return ChangelogMode.all();
     }
 
     @Override

diff --git a/...oracle-cdc/src/main/java/com/ververica/cdc/connectors/oracle/table/OracleTableSource.java b/...oracle-cdc/src/main/java/com/ververica/cdc/connectors/oracle/table/OracleTableSource.java
@@ -27,7 +27,6 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.types.RowKind;
 
 import com.ververica.cdc.connectors.base.options.StartupOptions;
 import com.ververica.cdc.connectors.base.source.jdbc.JdbcIncrementalSource;
@@ -138,12 +137,7 @@ public OracleTableSource(
 
     @Override
     public ChangelogMode getChangelogMode() {
-        return ChangelogMode.newBuilder()
-                .addContainedKind(RowKind.INSERT)
-                .addContainedKind(RowKind.UPDATE_BEFORE)
-                .addContainedKind(RowKind.UPDATE_AFTER)
-                .addContainedKind(RowKind.DELETE)
-                .build();
+        return ChangelogMode.all();
     }
 
     @Override

diff --git a/...-cdc/src/main/java/com/ververica/cdc/connectors/sqlserver/table/SqlServerTableSource.java b/...-cdc/src/main/java/com/ververica/cdc/connectors/sqlserver/table/SqlServerTableSource.java
@@ -26,7 +26,6 @@
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.RowType;
-import org.apache.flink.types.RowKind;
 
 import com.ververica.cdc.connectors.sqlserver.SqlServerSource;
 import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
@@ -102,12 +101,7 @@ public SqlServerTableSource(
 
     @Override
     public ChangelogMode getChangelogMode() {
-        return ChangelogMode.newBuilder()
-                .addContainedKind(RowKind.INSERT)
-                .addContainedKind(RowKind.UPDATE_BEFORE)
-                .addContainedKind(RowKind.UPDATE_AFTER)
-                .addContainedKind(RowKind.DELETE)
-                .build();
+        return ChangelogMode.all();
     }
 
     @Override