[cdc] Refactor mysql debezium json event parser (apache#2053)

liming30 · Sep 26, 2023 · 77f10a0 · 77f10a0
1 parent b6d9c92
commit 77f10a0
Show file tree

Hide file tree

Showing 17 changed files with 626 additions and 169 deletions.
diff --git a/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaUtils.java b/...on-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSchemaUtils.java
@@ -19,8 +19,8 @@
 package org.apache.paimon.flink.action.cdc.kafka;
 
 import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.formats.DataFormat;
-import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.DataFormat;
+import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
 import org.apache.paimon.schema.Schema;
 
 import org.apache.flink.configuration.Configuration;
@@ -44,7 +44,7 @@
 import java.util.stream.StreamSupport;
 
 import static org.apache.paimon.flink.action.cdc.kafka.KafkaActionUtils.kafkaPropertiesGroupId;
-import static org.apache.paimon.flink.action.cdc.kafka.formats.DataFormat.getDataFormat;
+import static org.apache.paimon.flink.action.cdc.kafka.format.DataFormat.getDataFormat;
 
 /** Utility class to load kafka schema. */
 public class KafkaSchemaUtils {

diff --git a/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java b/...k-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncDatabaseAction.java
@@ -24,8 +24,8 @@
 import org.apache.paimon.flink.action.MultiTablesSinkMode;
 import org.apache.paimon.flink.action.cdc.TableNameConverter;
 import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.formats.DataFormat;
-import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.DataFormat;
+import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
 import org.apache.paimon.flink.sink.cdc.EventParser;
 import org.apache.paimon.flink.sink.cdc.FlinkCdcSyncDatabaseSinkBuilder;
 import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;

diff --git a/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java b/...link-cdc/src/main/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableAction.java
@@ -26,8 +26,8 @@
 import org.apache.paimon.flink.action.ActionBase;
 import org.apache.paimon.flink.action.cdc.ComputedColumn;
 import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.formats.DataFormat;
-import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.DataFormat;
+import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
 import org.apache.paimon.flink.sink.cdc.CdcSinkBuilder;
 import org.apache.paimon.flink.sink.cdc.EventParser;
 import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;

diff --git a/.../action/cdc/kafka/formats/DataFormat.java → ...k/action/cdc/kafka/format/DataFormat.java b/.../action/cdc/kafka/formats/DataFormat.java → ...k/action/cdc/kafka/format/DataFormat.java
@@ -16,13 +16,13 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.action.cdc.kafka.formats;
+package org.apache.paimon.flink.action.cdc.kafka.format;
 
 import org.apache.paimon.flink.action.cdc.ComputedColumn;
 import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.formats.canal.CanalRecordParser;
-import org.apache.paimon.flink.action.cdc.kafka.formats.maxwell.MaxwellRecordParser;
-import org.apache.paimon.flink.action.cdc.kafka.formats.ogg.OggRecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.canal.CanalRecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.maxwell.MaxwellRecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.ogg.OggRecordParser;
 
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.connectors.kafka.table.KafkaConnectorOptions;

diff --git a/...ction/cdc/kafka/formats/RecordParser.java → ...action/cdc/kafka/format/RecordParser.java b/...ction/cdc/kafka/formats/RecordParser.java → ...action/cdc/kafka/format/RecordParser.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.action.cdc.kafka.formats;
+package org.apache.paimon.flink.action.cdc.kafka.format;
 
 import org.apache.paimon.flink.action.cdc.ComputedColumn;
 import org.apache.paimon.flink.action.cdc.TypeMapping;

diff --git a/...dc/kafka/formats/RecordParserFactory.java → ...cdc/kafka/format/RecordParserFactory.java b/...dc/kafka/formats/RecordParserFactory.java → ...cdc/kafka/format/RecordParserFactory.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.action.cdc.kafka.formats;
+package org.apache.paimon.flink.action.cdc.kafka.format;
 
 import org.apache.paimon.flink.action.cdc.ComputedColumn;
 import org.apache.paimon.flink.action.cdc.TypeMapping;

diff --git a/...kafka/formats/canal/CanalFieldParser.java → .../kafka/format/canal/CanalFieldParser.java b/...kafka/formats/canal/CanalFieldParser.java → .../kafka/format/canal/CanalFieldParser.java
@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.action.cdc.kafka.formats.canal;
+package org.apache.paimon.flink.action.cdc.kafka.format.canal;
 
 /** Converts some special types such as enum、set、geometry. */
 public class CanalFieldParser {

diff --git a/...afka/formats/canal/CanalRecordParser.java → ...kafka/format/canal/CanalRecordParser.java b/...afka/formats/canal/CanalRecordParser.java → ...kafka/format/canal/CanalRecordParser.java
@@ -16,11 +16,11 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.action.cdc.kafka.formats.canal;
+package org.apache.paimon.flink.action.cdc.kafka.format.canal;
 
 import org.apache.paimon.flink.action.cdc.ComputedColumn;
 import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
 import org.apache.paimon.flink.action.cdc.mysql.MySqlTypeUtils;
 import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
 import org.apache.paimon.types.DataType;

diff --git a/.../formats/maxwell/MaxwellRecordParser.java → ...a/format/maxwell/MaxwellRecordParser.java b/.../formats/maxwell/MaxwellRecordParser.java → ...a/format/maxwell/MaxwellRecordParser.java
@@ -16,11 +16,11 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.action.cdc.kafka.formats.maxwell;
+package org.apache.paimon.flink.action.cdc.kafka.format.maxwell;
 
 import org.apache.paimon.flink.action.cdc.ComputedColumn;
 import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
 import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
 import org.apache.paimon.types.RowKind;
 

diff --git a/...dc/kafka/formats/ogg/OggRecordParser.java → ...cdc/kafka/format/ogg/OggRecordParser.java b/...dc/kafka/formats/ogg/OggRecordParser.java → ...cdc/kafka/format/ogg/OggRecordParser.java
@@ -16,11 +16,11 @@
  * limitations under the License.
  */
 
-package org.apache.paimon.flink.action.cdc.kafka.formats.ogg;
+package org.apache.paimon.flink.action.cdc.kafka.format.ogg;
 
 import org.apache.paimon.flink.action.cdc.ComputedColumn;
 import org.apache.paimon.flink.action.cdc.TypeMapping;
-import org.apache.paimon.flink.action.cdc.kafka.formats.RecordParser;
+import org.apache.paimon.flink.action.cdc.kafka.format.RecordParser;
 import org.apache.paimon.flink.sink.cdc.RichCdcMultiplexRecord;
 import org.apache.paimon.types.RowKind;