fixup! [FLINK-36701][cdc-runtime] Add steps to get and emit schemaManager's latest evolvedSchema when SinkDataWriterOperator handles FlushEvent.

Author: Jzjsnow

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkFunctionOperator.java

@@ -112,6 +112,11 @@ public void processElement(StreamRecord<Event> element) throws Exception {
     // ----------------------------- Helper functions -------------------------------
     private void handleFlushEvent(FlushEvent event) throws Exception {
         userFunction.finish();
+        if (!processedTableIds.contains(event.getTableId()) && !event.getIsForCreateTableEvent()) {
+            LOG.info("Table {} has not been processed", event.getTableId());
+            emitLatestSchema(event.getTableId());
+            processedTableIds.add(event.getTableId());
+        }
         schemaEvolutionClient.notifyFlushSuccess(
                 getRuntimeContext().getIndexOfThisSubtask(), event.getTableId());
     }