fix startAsynchronouslySplit method enumeratorMetrics NPE

beryllw · beryllw · commit 6e7bb332b07f · 2025-02-18T18:00:49.000+08:00
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/HybridSplitAssigner.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/HybridSplitAssigner.java
@@ -83,7 +83,8 @@ public HybridSplitAssigner(
                         remainingTables,
                         isTableIdCaseSensitive,
                         dialect,
-                        offsetFactory),
+                        offsetFactory,
+                        enumeratorContext),
                 false,
                 sourceConfig.getSplitMetaGroupSize(),
                 offsetFactory,
@@ -104,7 +105,8 @@ public HybridSplitAssigner(
                         currentParallelism,
                         checkpoint.getSnapshotPendingSplits(),
                         dialect,
-                        offsetFactory),
+                        offsetFactory,
+                        enumeratorContext),
                 checkpoint.isStreamSplitAssigned(),
                 sourceConfig.getSplitMetaGroupSize(),
                 offsetFactory,
@@ -128,7 +130,7 @@ private HybridSplitAssigner(
 
     @Override
     public void open() {
-        this.enumeratorMetrics = new SourceEnumeratorMetrics(enumeratorContext.metricGroup());
+        this.enumeratorMetrics = snapshotSplitAssigner.getEnumeratorMetrics();
 
         if (isStreamSplitAssigned) {
             enumeratorMetrics.enterStreamReading();
@@ -137,8 +139,6 @@ public void open() {
         }
 
         snapshotSplitAssigner.open();
-        // init enumerator metrics
-        snapshotSplitAssigner.initEnumeratorMetrics(enumeratorMetrics);
     }
 
     @Override
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java
@@ -17,6 +17,8 @@
 
 package org.apache.flink.cdc.connectors.base.source.assigner;
 
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
 import org.apache.flink.cdc.connectors.base.config.SourceConfig;
 import org.apache.flink.cdc.connectors.base.dialect.DataSourceDialect;
 import org.apache.flink.cdc.connectors.base.source.assigner.splitter.ChunkSplitter;
@@ -104,7 +106,8 @@ public SnapshotSplitAssigner(
             List<TableId> remainingTables,
             boolean isTableIdCaseSensitive,
             DataSourceDialect<C> dialect,
-            OffsetFactory offsetFactory) {
+            OffsetFactory offsetFactory,
+            SplitEnumeratorContext<? extends SourceSplit> enumeratorContext) {
         this(
                 sourceConfig,
                 currentParallelism,
@@ -120,15 +123,17 @@ public SnapshotSplitAssigner(
                 dialect,
                 offsetFactory,
                 new ConcurrentHashMap<>(),
-                ChunkSplitterState.NO_SPLITTING_TABLE_STATE);
+                ChunkSplitterState.NO_SPLITTING_TABLE_STATE,
+                enumeratorContext);
     }
 
     public SnapshotSplitAssigner(
             C sourceConfig,
             int currentParallelism,
             SnapshotPendingSplitsState checkpoint,
             DataSourceDialect<C> dialect,
-            OffsetFactory offsetFactory) {
+            OffsetFactory offsetFactory,
+            SplitEnumeratorContext<? extends SourceSplit> enumeratorContext) {
         this(
                 sourceConfig,
                 currentParallelism,
@@ -144,7 +149,8 @@ public SnapshotSplitAssigner(
                 dialect,
                 offsetFactory,
                 new ConcurrentHashMap<>(),
-                checkpoint.getChunkSplitterState());
+                checkpoint.getChunkSplitterState(),
+                enumeratorContext);
     }
 
     private SnapshotSplitAssigner(
@@ -162,7 +168,8 @@ private SnapshotSplitAssigner(
             DataSourceDialect<C> dialect,
             OffsetFactory offsetFactory,
             Map<String, Long> splitFinishedCheckpointIds,
-            ChunkSplitterState chunkSplitterState) {
+            ChunkSplitterState chunkSplitterState,
+            SplitEnumeratorContext<? extends SourceSplit> enumeratorContext) {
         this.sourceConfig = sourceConfig;
         this.currentParallelism = currentParallelism;
         this.alreadyProcessedTables = alreadyProcessedTables;
@@ -188,13 +195,19 @@ private SnapshotSplitAssigner(
         this.offsetFactory = offsetFactory;
         this.splitFinishedCheckpointIds = splitFinishedCheckpointIds;
         chunkSplitter = createChunkSplitter(sourceConfig, dialect, chunkSplitterState);
+        this.enumeratorMetrics = new SourceEnumeratorMetrics(enumeratorContext.metricGroup());
+    }
+
+    public SourceEnumeratorMetrics getEnumeratorMetrics() {
+        return enumeratorMetrics;
     }
 
     @Override
     public void open() {
         chunkSplitter.open();
         discoveryCaptureTables();
         captureNewlyAddedTables();
+        initEnumeratorMetrics();
         startAsynchronouslySplit();
     }
 
@@ -295,10 +308,11 @@ private void captureNewlyAddedTables() {
         }
     }
 
-    /** This should be invoked after this class's open method. */
-    public void initEnumeratorMetrics(SourceEnumeratorMetrics enumeratorMetrics) {
-        this.enumeratorMetrics = enumeratorMetrics;
-
+    /**
+     * This should be invoked before this class's startAsynchronouslySplit method to avoid
+     * enumeratorMetrics NPE.
+     */
+    private void initEnumeratorMetrics() {
         this.enumeratorMetrics.enterSnapshotPhase();
         this.enumeratorMetrics.registerMetrics(
                 alreadyProcessedTables::size, assignedSplits::size, remainingSplits::size);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlHybridSplitAssigner.java
@@ -111,7 +111,7 @@ private MySqlHybridSplitAssigner(
 
     @Override
     public void open() {
-        this.enumeratorMetrics = new SourceEnumeratorMetrics(enumeratorContext.metricGroup());
+        this.enumeratorMetrics = snapshotSplitAssigner.getEnumeratorMetrics();
 
         if (isBinlogSplitAssigned) {
             enumeratorMetrics.enterStreamReading();
@@ -120,8 +120,6 @@ public void open() {
         }
 
         snapshotSplitAssigner.open();
-        // init enumerator metrics
-        snapshotSplitAssigner.initEnumeratorMetrics(enumeratorMetrics);
     }
 
     @Override
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java
@@ -187,6 +187,11 @@ private MySqlSnapshotSplitAssigner(
                 new MySqlPartition(sourceConfig.getMySqlConnectorConfig().getLogicalName());
         this.enumeratorContext = enumeratorContext;
         this.splitFinishedCheckpointIds = splitFinishedCheckpointIds;
+        this.enumeratorMetrics = new SourceEnumeratorMetrics(enumeratorContext.metricGroup());
+    }
+
+    public SourceEnumeratorMetrics getEnumeratorMetrics() {
+        return enumeratorMetrics;
     }
 
     @Override
@@ -195,6 +200,7 @@ public void open() {
         chunkSplitter.open();
         discoveryCaptureTables();
         captureNewlyAddedTables();
+        initEnumeratorMetrics();
         startAsynchronouslySplit();
     }
 
@@ -302,10 +308,11 @@ private void captureNewlyAddedTables() {
         }
     }
 
-    /** This should be invoked after this class's open method. */
-    public void initEnumeratorMetrics(SourceEnumeratorMetrics enumeratorMetrics) {
-        this.enumeratorMetrics = enumeratorMetrics;
-
+    /**
+     * This should be invoked before this class's startAsynchronouslySplit method to avoid
+     * enumeratorMetrics NPE.
+     */
+    private void initEnumeratorMetrics() {
         this.enumeratorMetrics.enterSnapshotPhase();
         this.enumeratorMetrics.registerMetrics(
                 alreadyProcessedTables::size, assignedSplits::size, remainingSplits::size);
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/fetch/PostgresScanFetchTaskTest.java
@@ -17,14 +17,14 @@
 
 package org.apache.flink.cdc.connectors.postgres.source.fetch;
 
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
 import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
 import org.apache.flink.cdc.connectors.base.dialect.JdbcDataSourceDialect;
 import org.apache.flink.cdc.connectors.base.source.assigner.SnapshotSplitAssigner;
 import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceRecords;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
-import org.apache.flink.cdc.connectors.base.source.metrics.SourceEnumeratorMetrics;
 import org.apache.flink.cdc.connectors.base.source.reader.external.AbstractScanFetchTask;
 import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
 import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher;
@@ -37,7 +37,6 @@
 import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffsetFactory;
 import org.apache.flink.cdc.connectors.postgres.testutils.RecordsFormatter;
 import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.types.DataType;
 
@@ -329,10 +328,8 @@ private List<SnapshotSplit> getSnapshotSplits(
                         discoverTables,
                         sourceDialect.isDataCollectionIdCaseSensitive(sourceConfig),
                         sourceDialect,
-                        offsetFactory);
-        snapshotSplitAssigner.initEnumeratorMetrics(
-                new SourceEnumeratorMetrics(
-                        UnregisteredMetricsGroup.createSplitEnumeratorMetricGroup()));
+                        offsetFactory,
+                        new MockSplitEnumeratorContext<>(1));
         snapshotSplitAssigner.open();
         List<SnapshotSplit> snapshotSplitList = new ArrayList<>();
         Optional<SourceSplitBase> split = snapshotSplitAssigner.getNext();
diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/read/fetch/SqlServerScanFetchTaskTest.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-sqlserver-cdc/src/test/java/org/apache/flink/cdc/connectors/sqlserver/source/read/fetch/SqlServerScanFetchTaskTest.java
@@ -17,14 +17,14 @@
 
 package org.apache.flink.cdc.connectors.sqlserver.source.read.fetch;
 
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
 import org.apache.flink.cdc.connectors.base.config.JdbcSourceConfig;
 import org.apache.flink.cdc.connectors.base.dialect.JdbcDataSourceDialect;
 import org.apache.flink.cdc.connectors.base.source.assigner.SnapshotSplitAssigner;
 import org.apache.flink.cdc.connectors.base.source.meta.offset.OffsetFactory;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SnapshotSplit;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceRecords;
 import org.apache.flink.cdc.connectors.base.source.meta.split.SourceSplitBase;
-import org.apache.flink.cdc.connectors.base.source.metrics.SourceEnumeratorMetrics;
 import org.apache.flink.cdc.connectors.base.source.reader.external.AbstractScanFetchTask;
 import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
 import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceScanFetcher;
@@ -37,7 +37,6 @@
 import org.apache.flink.cdc.connectors.sqlserver.source.reader.fetch.SqlServerScanFetchTask;
 import org.apache.flink.cdc.connectors.sqlserver.source.reader.fetch.SqlServerSourceFetchTaskContext;
 import org.apache.flink.cdc.connectors.sqlserver.testutils.RecordsFormatter;
-import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.table.api.DataTypes;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.RowType;
@@ -333,10 +332,8 @@ private List<SnapshotSplit> getSnapshotSplits(
                         discoverTables,
                         sourceDialect.isDataCollectionIdCaseSensitive(sourceConfig),
                         sourceDialect,
-                        offsetFactory);
-        snapshotSplitAssigner.initEnumeratorMetrics(
-                new SourceEnumeratorMetrics(
-                        UnregisteredMetricsGroup.createSplitEnumeratorMetricGroup()));
+                        offsetFactory,
+                        new MockSplitEnumeratorContext<>(1));
         snapshotSplitAssigner.open();
         List<SnapshotSplit> snapshotSplitList = new ArrayList<>();
         Optional<SourceSplitBase> split = snapshotSplitAssigner.getNext();

Original file line number	Diff line number	Diff line change
`@@ -111,7 +111,7 @@ private MySqlHybridSplitAssigner(`
`111`	`111`
`112`	`112`	`@Override`
`113`	`113`	`public void open() {`
`114`		`- this.enumeratorMetrics = new SourceEnumeratorMetrics(enumeratorContext.metricGroup());`
	`114`	`+ this.enumeratorMetrics = snapshotSplitAssigner.getEnumeratorMetrics();`
`115`	`115`
`116`	`116`	`if (isBinlogSplitAssigned) {`
`117`	`117`	`enumeratorMetrics.enterStreamReading();`
`@@ -120,8 +120,6 @@ public void open() {`
`120`	`120`	`}`
`121`	`121`
`122`	`122`	`snapshotSplitAssigner.open();`
`123`		`- // init enumerator metrics`
`124`		`- snapshotSplitAssigner.initEnumeratorMetrics(enumeratorMetrics);`
`125`	`123`	`}`
`126`	`124`
`127`	`125`	`@Override`