[HUDI-3233] Make metadata commit synchronous for flink batch

close apache/hudi#4561

hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java

@@ -233,7 +233,7 @@ public class StreamWriteOperatorCoordinator
             // start new instant.
             startInstant();
             // sync Hive if is enabled
-            syncHiveIfEnabled();
+            syncHiveAsync();
           }
         }, "commits the instant %s", this.instant
     );
@@ -246,21 +246,24 @@ public class StreamWriteOperatorCoordinator
 
   @Override
   public void handleEventFromOperator(int i, OperatorEvent operatorEvent) {
-    executor.execute(
+    ValidationUtils.checkState(operatorEvent instanceof WriteMetadataEvent,
-        () -> {
+        "The coordinator can only handle WriteMetaEvent");
-          // no event to handle
+    WriteMetadataEvent event = (WriteMetadataEvent) operatorEvent;
-          ValidationUtils.checkState(operatorEvent instanceof WriteMetadataEvent,
+
-              "The coordinator can only handle WriteMetaEvent");
+    if (event.isEndInput()) {
-          WriteMetadataEvent event = (WriteMetadataEvent) operatorEvent;
+      // handle end input event synchronously
-          if (event.isBootstrap()) {
+      handleEndInputEvent(event);
-            handleBootstrapEvent(event);
+    } else {
-          } else if (event.isEndInput()) {
+      executor.execute(
-            handleEndInputEvent(event);
+          () -> {
-          } else {
+            if (event.isBootstrap()) {
-            handleWriteMetaEvent(event);
+              handleBootstrapEvent(event);
-          }
+            } else {
-        }, "handle write metadata event for instant %s", this.instant
+              handleWriteMetaEvent(event);
-    );
+            }
+          }, "handle write metadata event for instant %s", this.instant
+      );
+    }
   }
 
   @Override
@@ -289,16 +292,23 @@ public class StreamWriteOperatorCoordinator
     this.hiveSyncContext = HiveSyncContext.create(conf);
   }
 
-  private void syncHiveIfEnabled() {
+  private void syncHiveAsync() {
     if (tableState.syncHive) {
-      this.hiveSyncExecutor.execute(this::syncHive, "sync hive metadata for instant %s", this.instant);
+      this.hiveSyncExecutor.execute(this::doSyncHive, "sync hive metadata for instant %s", this.instant);
+    }
+  }
+
+  private void syncHive() {
+    if (tableState.syncHive) {
+      doSyncHive();
+      LOG.info("Sync hive metadata for instant {} success!", this.instant);
     }
   }
 
   /**
    * Sync hoodie table metadata to Hive metastore.
    */
-  public void syncHive() {
+  public void doSyncHive() {
     hiveSyncContext.hiveSyncTool().syncHoodieTable();
   }
 
@@ -380,8 +390,8 @@ public class StreamWriteOperatorCoordinator
       // The executor thread inherits the classloader of the #handleEventFromOperator
       // caller, which is a AppClassLoader.
       Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
-      // sync Hive if is enabled in batch mode.
+      // sync Hive synchronously if it is enabled in batch mode.
-      syncHiveIfEnabled();
+      syncHive();
     }
   }