[FLINK-36742][cdc-base] Filter unacked split for no capture tables when task restore from state

This closes  #3739

Author: gong

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/reader/IncrementalSourceReader.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.api.connector.source.SourceEvent;
 import org.apache.flink.cdc.common.annotation.Experimental;
+import org.apache.flink.cdc.common.annotation.VisibleForTesting;
 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.meta.events.FinishedSnapshotSplitsAckEvent;
@@ -261,13 +262,14 @@ private void addSplits(List<SourceSplitBase> splits, boolean checkTableChangeFor
         for (SourceSplitBase split : splits) {
             if (split.isSnapshotSplit()) {
                 SnapshotSplit snapshotSplit = split.asSnapshotSplit();
-                if (snapshotSplit.isSnapshotReadFinished()) {
-                    finishedUnackedSplits.put(snapshotSplit.splitId(), snapshotSplit);
-                } else if (dialect.isIncludeDataCollection(
-                        sourceConfig, snapshotSplit.getTableId())) {
-                    unfinishedSplits.add(split);
+                if (dialect.isIncludeDataCollection(sourceConfig, snapshotSplit.getTableId())) {
+                    if (snapshotSplit.isSnapshotReadFinished()) {
+                        finishedUnackedSplits.put(snapshotSplit.splitId(), snapshotSplit);
+                    } else {
+                        unfinishedSplits.add(split);
+                    }
                 } else {
-                    LOG.debug(
+                    LOG.info(
                             "The subtask {} is skipping split {} because it does not match new table filter.",
                             subtaskId,
                             split.splitId());
@@ -320,8 +322,9 @@ private void addSplits(List<SourceSplitBase> splits, boolean checkTableChangeFor
         // add all un-finished splits (including binlog split) to SourceReaderBase
         if (!unfinishedSplits.isEmpty()) {
             super.addSplits(unfinishedSplits);
-        } else if (suspendedStreamSplit
-                != null) { // only request new snapshot split if the stream split is suspended
+        } else if (suspendedStreamSplit != null
+                || getNumberOfCurrentlyAssignedSplits()
+                        <= 1) { // only request new snapshot split if the stream split is suspended
             context.sendSplitRequest();
         }
     }
@@ -541,4 +544,9 @@ private void logCurrentStreamOffsets(List<SourceSplitBase> splits, long checkpoi
             LOG.info("Stream split offset on checkpoint {}: {}", checkpointId, offset);
         }
     }
+
+    @VisibleForTesting
+    public Map<String, SnapshotSplit> getFinishedUnackedSplits() {
+        return finishedUnackedSplits;
+    }
 }

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-oracle-cdc/pom.xml

@@ -164,6 +164,12 @@ limitations under the License.
             <artifactId>commons-lang3</artifactId>
             <version>${commons-lang3.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-connector-test-utils</artifactId>
+            <version>${flink.version}</version>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>