[FLINK-37110][cdc-runtime] Fixed the problem of using the same executor to process schema change requests and flush event requests, resulting in blocking timeout.

Author: linjc13

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaCoordinator.java

@@ -61,6 +61,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
@@ -98,6 +99,9 @@ public class SchemaCoordinator extends SchemaRegistry {
     private transient Multimap<Tuple2<Integer, SchemaChangeEvent>, Integer>
             alreadyHandledSchemaChangeEvents;
 
+    /** Executor service to execute schema change. */
+    private final ExecutorService schemaChangeThreadPool;
+
     public SchemaCoordinator(
             String operatorName,
             OperatorCoordinator.Context context,
@@ -114,6 +118,7 @@ public SchemaCoordinator(
                 routingRules,
                 schemaChangeBehavior,
                 rpcTimeout);
+        this.schemaChangeThreadPool = Executors.newSingleThreadExecutor();
     }
 
     // -----------------
@@ -131,6 +136,14 @@ public void start() throws Exception {
                 "Started SchemaRegistry for {}. Parallelism: {}", operatorName, currentParallelism);
     }
 
+    @Override
+    public void close() throws Exception {
+        super.close();
+        if (schemaChangeThreadPool != null && !schemaChangeThreadPool.isShutdown()) {
+            schemaChangeThreadPool.shutdownNow();
+        }
+    }
+
     // --------------------------
     // Checkpoint related methods
     // --------------------------
@@ -268,7 +281,20 @@ private void handleSchemaEvolveRequest(
             LOG.info(
                     "Received the last required schema change request {}. Switching from WAITING_FOR_FLUSH to EVOLVING.",
                     request);
-            startSchemaChange();
+
+            schemaChangeThreadPool.submit(
+                    () -> {
+                        try {
+                            startSchemaChange();
+                        } catch (Throwable t) {
+                            failJob(
+                                    "Schema change applying task",
+                                    new FlinkRuntimeException(
+                                            "Failed to apply schema change event.", t));
+                            throw new FlinkRuntimeException(
+                                    "Failed to apply schema change event.", t);
+                        }
+                    });
         }
     }