[cdc-runtime][hotfix] Setup waitFlushSuccess before responding operat…

…or to flush. (#2805)

This closes #2805.

There might be a case that sink flushes faster than schema operator requesting release upstream. In that case waitFlushSuccess has not been set when sink has flushed successfully, which will lead to a NullPointerException.

flink-cdc-runtime/src/main/java/com/ververica/cdc/runtime/operators/schema/coordinator/SchemaRegistryRequestHandler.java

@@ -104,7 +104,8 @@ public CompletableFuture<CoordinationResponse> handleSchemaChangeRequest(
             CompletableFuture<CoordinationResponse> response =
                     CompletableFuture.completedFuture(wrap(new SchemaChangeResponse(true)));
             schemaManager.applySchemaChange(request.getSchemaChangeEvent());
-            pendingSchemaChanges.add(new PendingSchemaChange(request, response));
+            this.waitFlushSuccess =
+                    new PendingSchemaChange(request, response).startToWaitForFlushSuccess();
             return response;
         } else {
             LOG.info("There are already processing requests. Wait for processing.");
@@ -116,7 +117,6 @@ public CompletableFuture<CoordinationResponse> handleSchemaChangeRequest(
 
     /** Handle the {@link ReleaseUpstreamRequest} and wait for all sink subtasks flushing. */
     public CompletableFuture<CoordinationResponse> handleReleaseUpstreamRequest() {
-        this.waitFlushSuccess = pendingSchemaChanges.remove(0).startToWaitForFlushSuccess();
         return waitFlushSuccess.getResponseFuture();
     }
 
@@ -152,19 +152,19 @@ private void startNextSchemaChangeRequest() {
         flushedSinkWriters.clear();
         waitFlushSuccess = null;
         while (!pendingSchemaChanges.isEmpty()) {
-            PendingSchemaChange pendingSchemaChange = pendingSchemaChanges.get(0);
+            PendingSchemaChange pendingSchemaChange = pendingSchemaChanges.remove(0);
             SchemaChangeRequest request = pendingSchemaChange.changeRequest;
             if (request.getSchemaChangeEvent() instanceof CreateTableEvent
                     && schemaManager.schemaExists(request.getTableId())) {
                 pendingSchemaChange
                         .getResponseFuture()
                         .complete(wrap(new SchemaChangeResponse(false)));
-                pendingSchemaChanges.remove(0);
             } else {
                 schemaManager.applySchemaChange(request.getSchemaChangeEvent());
                 pendingSchemaChange
                         .getResponseFuture()
                         .complete(wrap(new SchemaChangeResponse(true)));
+                this.waitFlushSuccess = pendingSchemaChange.startToWaitForFlushSuccess();
                 break;
             }
         }