[FLINK-36771][cdc-base][mysql] Fix UT trigger error: Invalid assigner status {} [NEWLY_ADDED_ASSIGNING_FINISHED]

Author: gong

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/enumerator/IncrementalSourceEnumerator.java

@@ -275,17 +275,18 @@ protected void syncWithReaders(int[] subtaskIds, Throwable t) {
     private void requestStreamSplitUpdateIfNeed() {
         if (!isStreamSplitUpdateRequestAlreadySent
                 && isNewlyAddedAssigningSnapshotFinished(splitAssigner.getAssignerStatus())) {
-            isStreamSplitUpdateRequestAlreadySent = true;
             // If enumerator knows which reader is assigned stream split, just send to this reader,
             // nor sends to all registered readers.
             if (streamSplitTaskId != null) {
+                isStreamSplitUpdateRequestAlreadySent = true;
                 LOG.info(
                         "The enumerator requests subtask {} to update the stream split after newly added table.",
                         streamSplitTaskId);
                 context.sendEventToSourceReader(
                         streamSplitTaskId, new StreamSplitUpdateRequestEvent());
             } else {
                 for (int reader : getRegisteredReader()) {
+                    isStreamSplitUpdateRequestAlreadySent = true;
                     LOG.info(
                             "The enumerator requests subtask {} to update the stream split after newly added table.",
                             reader);

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/enumerator/MySqlSourceEnumerator.java

@@ -277,8 +277,8 @@ private void syncWithReaders(int[] subtaskIds, Throwable t) {
     private void requestBinlogSplitUpdateIfNeed() {
         if (!isBinlogSplitUpdateRequestAlreadySent
                 && isNewlyAddedAssigningSnapshotFinished(splitAssigner.getAssignerStatus())) {
-            isBinlogSplitUpdateRequestAlreadySent = true;
             for (int subtaskId : getRegisteredReader()) {
+                isBinlogSplitUpdateRequestAlreadySent = true;
                 LOG.info(
                         "The enumerator requests subtask {} to update the binlog split after newly added table.",
                         subtaskId);