[cdc-core] Fix BaseSchema and PooledDataSourceFactory classes Serialization exception

Author: yexianxun

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/assigner/HybridSplitAssigner.java

@@ -18,8 +18,8 @@
 
 package com.ververica.cdc.connectors.core.assigner;
 
-import com.ververica.cdc.connectors.core.assigner.state.PendingSplitsState;
 import com.ververica.cdc.connectors.core.assigner.state.HybridPendingSplitsState;
+import com.ververica.cdc.connectors.core.assigner.state.PendingSplitsState;
 import com.ververica.cdc.connectors.core.config.SourceConfig;
 import com.ververica.cdc.connectors.core.dialect.SnapshotEventDialect;
 import com.ververica.cdc.connectors.core.offset.Offset;
@@ -63,7 +63,7 @@ public HybridSplitAssigner(
             boolean isTableIdCaseSensitive,
             SnapshotEventDialect dialect,
             OffsetFactory offsetFactory,
-            BaseSchema baseSchema) {
+            BaseSchema.SchemaFactory schemaFactory) {
         this(
                 new SnapshotSplitAssigner(
                         sourceConfig,
@@ -72,7 +72,7 @@ public HybridSplitAssigner(
                         isTableIdCaseSensitive,
                         dialect,
                         offsetFactory,
-                        baseSchema),
+                        schemaFactory),
                 false,
                 sourceConfig.getSplitMetaGroupSize());
         this.offsetFactory = offsetFactory;
@@ -84,15 +84,15 @@ public HybridSplitAssigner(
             HybridPendingSplitsState checkpoint,
             SnapshotEventDialect dialect,
             OffsetFactory offsetFactory,
-            BaseSchema baseSchema) {
+            BaseSchema.SchemaFactory schemaFactory) {
         this(
                 new SnapshotSplitAssigner(
                         sourceConfig,
                         currentParallelism,
                         checkpoint.getSnapshotPendingSplits(),
                         dialect,
                         offsetFactory,
-                        baseSchema),
+                        schemaFactory),
                 checkpoint.isStreamSplitAssigned(),
                 sourceConfig.getSplitMetaGroupSize());
     }

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/assigner/SnapshotSplitAssigner.java

@@ -67,7 +67,7 @@ public class SnapshotSplitAssigner implements SplitAssigner {
     @Nullable private Long checkpointIdToFinish;
     private final SnapshotEventDialect dialect;
     private OffsetFactory offsetFactory;
-    private final BaseSchema baseSchema;
+    private final BaseSchema.SchemaFactory schemaFactory;
 
     public SnapshotSplitAssigner(
             SourceConfig sourceConfig,
@@ -76,7 +76,7 @@ public SnapshotSplitAssigner(
             boolean isTableIdCaseSensitive,
             SnapshotEventDialect dialect,
             OffsetFactory offsetFactory,
-            BaseSchema baseSchema) {
+            BaseSchema.SchemaFactory schemaFactory) {
         this(
                 sourceConfig,
                 currentParallelism,
@@ -90,7 +90,7 @@ public SnapshotSplitAssigner(
                 true,
                 dialect,
                 offsetFactory,
-                baseSchema);
+                schemaFactory);
     }
 
     public SnapshotSplitAssigner(
@@ -99,7 +99,7 @@ public SnapshotSplitAssigner(
             SnapshotPendingSplitsState checkpoint,
             SnapshotEventDialect dialect,
             OffsetFactory offsetFactory,
-            BaseSchema baseSchema) {
+            BaseSchema.SchemaFactory schemaFactory) {
         this(
                 sourceConfig,
                 currentParallelism,
@@ -113,7 +113,7 @@ public SnapshotSplitAssigner(
                 checkpoint.isRemainingTablesCheckpointed(),
                 dialect,
                 offsetFactory,
-                baseSchema);
+                schemaFactory);
     }
 
     private SnapshotSplitAssigner(
@@ -129,7 +129,7 @@ private SnapshotSplitAssigner(
             boolean isRemainingTablesCheckpointed,
             SnapshotEventDialect dialect,
             OffsetFactory offsetFactory,
-            BaseSchema baseSchema) {
+            BaseSchema.SchemaFactory schemaFactory) {
         this.sourceConfig = sourceConfig;
         this.currentParallelism = currentParallelism;
         this.alreadyProcessedTables = alreadyProcessedTables;
@@ -142,7 +142,7 @@ private SnapshotSplitAssigner(
         this.isTableIdCaseSensitive = isTableIdCaseSensitive;
         this.dialect = dialect;
         this.offsetFactory = offsetFactory;
-        this.baseSchema = baseSchema;
+        this.schemaFactory = schemaFactory;
     }
 
     @Override
@@ -313,6 +313,7 @@ private boolean allSplitsFinished() {
     }
 
     private ChunkSplitter createChunkSplitter(SourceConfig sourceConfig) {
+        BaseSchema baseSchema = schemaFactory.create(sourceConfig, isTableIdCaseSensitive);
         return new ChunkSplitter(baseSchema, sourceConfig, dialect);
     }
 }

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/assigner/StreamSplitAssigner.java

@@ -20,9 +20,9 @@
 
 import com.ververica.cdc.connectors.core.assigner.state.PendingSplitsState;
 import com.ververica.cdc.connectors.core.assigner.state.StreamPendingSplitsState;
-import com.ververica.cdc.connectors.core.internal.connection.JdbcConnectionFactory;
 import com.ververica.cdc.connectors.core.config.SourceConfig;
 import com.ververica.cdc.connectors.core.dialect.Dialect;
+import com.ververica.cdc.connectors.core.internal.connection.JdbcConnectionFactory;
 import com.ververica.cdc.connectors.core.offset.Offset;
 import com.ververica.cdc.connectors.core.offset.OffsetFactory;
 import com.ververica.cdc.connectors.core.split.FinishedSnapshotSplitInfo;

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/dialect/Dialect.java

@@ -24,10 +24,10 @@
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.flink.util.FlinkRuntimeException;
 
+import com.ververica.cdc.connectors.core.config.SourceConfig;
 import com.ververica.cdc.connectors.core.internal.connection.JdbcConnectionFactory;
 import com.ververica.cdc.connectors.core.internal.connection.PooledDataSourceFactory;
 import com.ververica.cdc.connectors.core.internal.converter.JdbcSourceRecordConverter;
-import com.ververica.cdc.connectors.core.config.SourceConfig;
 import com.ververica.cdc.connectors.core.offset.Offset;
 import io.debezium.connector.base.ChangeEventQueue;
 import io.debezium.jdbc.JdbcConnection;

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/dialect/SnapshotEventDialect.java

@@ -20,11 +20,11 @@
 
 import org.apache.flink.table.types.logical.RowType;
 
+import com.ververica.cdc.connectors.core.config.SourceConfig;
 import com.ververica.cdc.connectors.core.internal.converter.JdbcSourceRecordConverter;
+import com.ververica.cdc.connectors.core.offset.Offset;
 import com.ververica.cdc.connectors.core.reader.split.SnapshotReader;
 import com.ververica.cdc.connectors.core.reader.split.dispatcher.SignalEventDispatcher;
-import com.ververica.cdc.connectors.core.config.SourceConfig;
-import com.ververica.cdc.connectors.core.offset.Offset;
 import com.ververica.cdc.connectors.core.split.SnapshotSplit;
 import com.ververica.cdc.connectors.core.split.SourceSplitBase;
 import io.debezium.DebeziumException;

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/internal/connection/PooledDataSourceFactory.java

@@ -22,8 +22,11 @@
 import com.zaxxer.hikari.HikariConfig;
 import com.zaxxer.hikari.HikariDataSource;
 
+import java.io.Serializable;
+
 /** A connection pool factory to create pooled DataSource {@link HikariDataSource}. */
-public abstract class PooledDataSourceFactory {
+public abstract class PooledDataSourceFactory implements Serializable {
+    private static final long serialVersionUID = 1L;
 
     public static final String CONNECTION_POOL_PREFIX = "connection-pool-";
     public static final String SERVER_TIMEZONE_KEY = "serverTimezone";

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/reader/BaseSplitReader.java

@@ -23,12 +23,12 @@
 import org.apache.flink.connector.base.source.reader.splitreader.SplitsAddition;
 import org.apache.flink.connector.base.source.reader.splitreader.SplitsChange;
 
+import com.ververica.cdc.connectors.core.config.SourceConfig;
 import com.ververica.cdc.connectors.core.dialect.SnapshotEventDialect;
 import com.ververica.cdc.connectors.core.dialect.StreamingEventDialect;
 import com.ververica.cdc.connectors.core.reader.split.Reader;
 import com.ververica.cdc.connectors.core.reader.split.SnapshotReader;
 import com.ververica.cdc.connectors.core.reader.split.StreamReader;
-import com.ververica.cdc.connectors.core.config.SourceConfig;
 import com.ververica.cdc.connectors.core.split.ChangeEventRecords;
 import com.ververica.cdc.connectors.core.split.SourceSplitBase;
 import org.apache.kafka.connect.source.SourceRecord;

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/schema/BaseSchema.java

@@ -18,10 +18,13 @@
 
 package com.ververica.cdc.connectors.core.schema;
 
+import com.ververica.cdc.connectors.core.config.SourceConfig;
 import io.debezium.jdbc.JdbcConnection;
 import io.debezium.relational.TableId;
 import io.debezium.relational.history.TableChanges.TableChange;
 
+import java.io.Serializable;
+
 /** Provides as a tool class to obtain table schema information. */
 public interface BaseSchema {
 
@@ -33,4 +36,9 @@ public interface BaseSchema {
      * @return An abstract representation of the structure to the tables of a relational database.
      */
     TableChange getTableSchema(JdbcConnection jdbc, TableId tableId);
+
+    /** A Schema Factory create a {@link BaseSchema}. */
+    interface SchemaFactory extends Serializable {
+        BaseSchema create(SourceConfig sourceConfig, boolean isTableIdCaseSensitive);
+    }
 }

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/source/ChangeEventHybridSource.java

@@ -34,16 +34,16 @@
 import com.ververica.cdc.connectors.core.assigner.HybridSplitAssigner;
 import com.ververica.cdc.connectors.core.assigner.SplitAssigner;
 import com.ververica.cdc.connectors.core.assigner.StreamSplitAssigner;
+import com.ververica.cdc.connectors.core.assigner.state.HybridPendingSplitsState;
 import com.ververica.cdc.connectors.core.assigner.state.PendingSplitsState;
 import com.ververica.cdc.connectors.core.assigner.state.PendingSplitsStateSerializer;
 import com.ververica.cdc.connectors.core.assigner.state.StreamPendingSplitsState;
-import com.ververica.cdc.connectors.core.dialect.StreamingEventDialect;
-import com.ververica.cdc.connectors.core.enumerator.SourceEnumerator;
-import com.ververica.cdc.connectors.core.assigner.state.HybridPendingSplitsState;
 import com.ververica.cdc.connectors.core.config.SourceConfig;
 import com.ververica.cdc.connectors.core.config.SourceConfigFactory;
 import com.ververica.cdc.connectors.core.config.StartupMode;
 import com.ververica.cdc.connectors.core.dialect.SnapshotEventDialect;
+import com.ververica.cdc.connectors.core.dialect.StreamingEventDialect;
+import com.ververica.cdc.connectors.core.enumerator.SourceEnumerator;
 import com.ververica.cdc.connectors.core.metrics.SourceReaderMetrics;
 import com.ververica.cdc.connectors.core.offset.OffsetFactory;
 import com.ververica.cdc.connectors.core.reader.BaseRecordEmitter;
@@ -76,23 +76,23 @@ public class ChangeEventHybridSource<T>
     private final Validator validator;
     private final SnapshotEventDialect snapshotEventDialect;
     private final StreamingEventDialect streamingEventDialect;
-    private final BaseSchema baseSchema;
+    private final BaseSchema.SchemaFactory schemaFactory;
 
     public ChangeEventHybridSource(
             SourceConfigFactory configFactory,
             DebeziumDeserializationSchema<T> deserializationSchema,
             OffsetFactory offsetFactory,
             SnapshotEventDialect snapshotEventDialect,
             StreamingEventDialect streamingEventDialect,
-            BaseSchema baseSchema) {
+            BaseSchema.SchemaFactory schemaFactory) {
         this(
                 configFactory,
                 deserializationSchema,
                 offsetFactory,
                 snapshotEventDialect,
                 streamingEventDialect,
                 Validator.getDefaultValidator(),
-                baseSchema);
+                schemaFactory);
     }
 
     public ChangeEventHybridSource(
@@ -102,7 +102,7 @@ public ChangeEventHybridSource(
             SnapshotEventDialect snapshotEventDialect,
             StreamingEventDialect streamingEventDialect,
             Validator validator,
-            BaseSchema baseSchema) {
+            BaseSchema.SchemaFactory schemaFactory) {
         this.configFactory = configFactory;
         this.deserializationSchema = deserializationSchema;
         this.offsetFactory = offsetFactory;
@@ -116,7 +116,7 @@ public OffsetFactory getOffsetFactory() {
                     }
                 };
         this.validator = validator;
-        this.baseSchema = baseSchema;
+        this.schemaFactory = schemaFactory;
     }
 
     @Override
@@ -177,7 +177,7 @@ public SplitEnumerator<SourceSplitBase, PendingSplitsState> createEnumerator(
                                 isTableIdCaseSensitive,
                                 snapshotEventDialect,
                                 offsetFactory,
-                                baseSchema);
+                                schemaFactory);
             } catch (Exception e) {
                 throw new FlinkRuntimeException(
                         "Failed to discover captured tables for enumerator", e);
@@ -204,7 +204,7 @@ public SplitEnumerator<SourceSplitBase, PendingSplitsState> restoreEnumerator(
                             (HybridPendingSplitsState) checkpoint,
                             snapshotEventDialect,
                             offsetFactory,
-                            baseSchema);
+                            schemaFactory);
         } else if (checkpoint instanceof StreamPendingSplitsState) {
             splitAssigner =
                     new StreamSplitAssigner(

flink-cdc-core/src/main/java/com/ververica/cdc/connectors/core/split/SourceSplitSerializer.java

@@ -42,8 +42,6 @@
 import java.util.List;
 import java.util.Map;
 
-import static com.ververica.cdc.connectors.core.utils.SerializerUtils.rowToSerializedString;
-
 /** A serializer for the {@link SourceSplitBase}. */
 public abstract class SourceSplitSerializer
         implements SimpleVersionedSerializer<SourceSplitBase>, OffsetDeserializerSerializer {