[HUDI-9381] Async instant time generation of Flink writers

The new instant time generation utilizes RPC request to coordinate creation of new instants.
Each write task will send an RPC request to the coordinator for the instant time, the coordinator
uses a global lock to guard the access from multiple tasks.

Now one checkpoint id corresponds to one instant.

Basic work flow:

* write task send request: current ckp id
* write task expected response: the instant time
* coordinator mappings of checkpoint and instant: ckp-id → {instant → {write-task-id → write meta event}}

Note that the ckp id used in the request is the last known id instead of the current checkpoint, for job restored
from a state, it is the state checkpoint id, otherwise -1 for fresh new job.

Author: danny0405

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/embedded/EmbeddedTimelineService.java

@@ -170,12 +170,6 @@ private void startServer(TimelineServiceCreator timelineServiceCreator) throws I
                   * writeConfig.getHoodieClientHeartbeatTolerableMisses());
     }
 
-    if (writeConfig.isTimelineServerBasedInstantStateEnabled()) {
-      timelineServiceConfBuilder
-          .instantStateForceRefreshRequestNumber(writeConfig.getTimelineServerBasedInstantStateForceRefreshRequestNumber())
-          .enableInstantStateRequests(true);
-    }
-
     this.serviceConfig = timelineServiceConfBuilder.build();
 
     server = timelineServiceCreator.create(context, storageConf.newInstance(), serviceConfig, viewManager);
@@ -191,10 +185,10 @@ TimelineService create(HoodieEngineContext context, StorageConfiguration<?> stor
 
   private void setHostAddr(String embeddedTimelineServiceHostAddr) {
     if (embeddedTimelineServiceHostAddr != null) {
-      LOG.info("Overriding hostIp to (" + embeddedTimelineServiceHostAddr + ") found in spark-conf. It was " + this.hostAddr);
+      LOG.info("Overriding hostIp to (" + embeddedTimelineServiceHostAddr + ") found in write conf. It was " + this.hostAddr);
       this.hostAddr = embeddedTimelineServiceHostAddr;
     } else {
-      LOG.warn("Unable to find driver bind address from spark config");
+      LOG.warn("Unable to find driver bind address from write config, use current host name");
       this.hostAddr = NetworkUtils.getHostname();
     }
   }
@@ -258,23 +252,23 @@ public void stopForBasePath(String basePath) {
 
   private static TimelineServiceIdentifier getTimelineServiceIdentifier(String hostAddr, HoodieWriteConfig writeConfig) {
     return new TimelineServiceIdentifier(hostAddr, writeConfig.getMarkersType(), writeConfig.isMetadataTableEnabled(),
-        writeConfig.isEarlyConflictDetectionEnable(), writeConfig.isTimelineServerBasedInstantStateEnabled());
+        writeConfig.isEarlyConflictDetectionEnable());
   }
 
   static class TimelineServiceIdentifier {
     private final String hostAddr;
     private final MarkerType markerType;
     private final boolean isMetadataEnabled;
     private final boolean isEarlyConflictDetectionEnable;
-    private final boolean isTimelineServerBasedInstantStateEnabled;
 
-    public TimelineServiceIdentifier(String hostAddr, MarkerType markerType, boolean isMetadataEnabled, boolean isEarlyConflictDetectionEnable,
-                                     boolean isTimelineServerBasedInstantStateEnabled) {
+    public TimelineServiceIdentifier(String hostAddr,
+                                     MarkerType markerType,
+                                     boolean isMetadataEnabled,
+                                     boolean isEarlyConflictDetectionEnable) {
       this.hostAddr = hostAddr;
       this.markerType = markerType;
       this.isMetadataEnabled = isMetadataEnabled;
       this.isEarlyConflictDetectionEnable = isEarlyConflictDetectionEnable;
-      this.isTimelineServerBasedInstantStateEnabled = isTimelineServerBasedInstantStateEnabled;
     }
 
     @Override
@@ -288,15 +282,15 @@ public boolean equals(Object o) {
       TimelineServiceIdentifier that = (TimelineServiceIdentifier) o;
       if (this.hostAddr != null && that.hostAddr != null) {
         return isMetadataEnabled == that.isMetadataEnabled && isEarlyConflictDetectionEnable == that.isEarlyConflictDetectionEnable
-            && isTimelineServerBasedInstantStateEnabled == that.isTimelineServerBasedInstantStateEnabled && hostAddr.equals(that.hostAddr) && markerType == that.markerType;
+            && hostAddr.equals(that.hostAddr) && markerType == that.markerType;
       } else {
         return (hostAddr == null && that.hostAddr == null);
       }
     }
 
     @Override
     public int hashCode() {
-      return Objects.hash(hostAddr, markerType, isMetadataEnabled, isEarlyConflictDetectionEnable, isTimelineServerBasedInstantStateEnabled);
+      return Objects.hash(hostAddr, markerType, isMetadataEnabled, isEarlyConflictDetectionEnable);
     }
   }
 }

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/transaction/lock/FileSystemBasedLockProvider.java

@@ -52,6 +52,7 @@
 
 import static org.apache.hudi.common.config.LockConfiguration.FILESYSTEM_LOCK_EXPIRE_PROP_KEY;
 import static org.apache.hudi.common.config.LockConfiguration.FILESYSTEM_LOCK_PATH_PROP_KEY;
+import static org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY;
 import static org.apache.hudi.common.table.HoodieTableMetaClient.AUXILIARYFOLDER_NAME;
 
 /**
@@ -208,6 +209,7 @@ public static TypedProperties getLockConfig(String tablePath) {
     TypedProperties props = new TypedProperties();
     props.put(HoodieLockConfig.LOCK_PROVIDER_CLASS_NAME.key(), FileSystemBasedLockProvider.class.getName());
     props.put(HoodieLockConfig.LOCK_ACQUIRE_WAIT_TIMEOUT_MS.key(), "2000");
+    props.put(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, "200");
     props.put(HoodieLockConfig.FILESYSTEM_LOCK_EXPIRE.key(), "1");
     props.put(HoodieLockConfig.LOCK_ACQUIRE_CLIENT_NUM_RETRIES.key(), "30");
     props.put(HoodieLockConfig.FILESYSTEM_LOCK_PATH.key(), defaultLockPath(tablePath));

hudi-client/hudi-client-common/src/main/java/org/apache/hudi/config/HoodieWriteConfig.java

@@ -506,18 +506,6 @@ public class HoodieWriteConfig extends HoodieConfig {
       .sinceVersion("0.9.0")
       .withDocumentation(MarkerType.class);
 
-  public static final ConfigProperty<Boolean> INSTANT_STATE_TIMELINE_SERVER_BASED = ConfigProperty
-      .key("hoodie.instant_state.timeline_server_based.enabled")
-      .defaultValue(false)
-      .sinceVersion("1.0.0")
-      .withDocumentation("If enabled, writers get instant state from timeline server rather than requesting DFS directly");
-
-  public static final ConfigProperty<Integer> INSTANT_STATE_TIMELINE_SERVER_BASED_FORCE_REFRESH_REQUEST_NUMBER = ConfigProperty
-      .key("hoodie.instant_state.timeline_server_based.force_refresh.request.number")
-      .defaultValue(100)
-      .sinceVersion("1.0.0")
-      .withDocumentation("Number of requests to trigger instant state cache refreshing");
-
   public static final ConfigProperty<Integer> MARKERS_TIMELINE_SERVER_BASED_BATCH_NUM_THREADS = ConfigProperty
       .key("hoodie.markers.timeline_server_based.batch.num_threads")
       .defaultValue(20)
@@ -1526,14 +1514,6 @@ public int getFinalizeWriteParallelism() {
     return getInt(FINALIZE_WRITE_PARALLELISM_VALUE);
   }
 
-  public boolean isTimelineServerBasedInstantStateEnabled() {
-    return getBoolean(INSTANT_STATE_TIMELINE_SERVER_BASED);
-  }
-
-  public int getTimelineServerBasedInstantStateForceRefreshRequestNumber() {
-    return getInt(INSTANT_STATE_TIMELINE_SERVER_BASED_FORCE_REFRESH_REQUEST_NUMBER);
-  }
-
   public MarkerType getMarkersType() {
     String markerType = getString(MARKERS_TYPE);
     return MarkerType.valueOf(markerType.toUpperCase());
@@ -3366,11 +3346,6 @@ public Builder withEmbeddedTimelineServerPort(int port) {
       return this;
     }
 
-    public Builder withTimelineServerBasedInstantStateEnable(boolean enable) {
-      writeConfig.setValue(INSTANT_STATE_TIMELINE_SERVER_BASED, String.valueOf(enable));
-      return this;
-    }
-
     public Builder withBulkInsertSortMode(String mode) {
       writeConfig.setValue(BULK_INSERT_SORT_MODE, mode);
       return this;

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/TimeGeneratorBase.java

@@ -33,7 +33,6 @@
 import java.util.Arrays;
 import java.util.concurrent.TimeUnit;
 
-import static org.apache.hudi.common.config.LockConfiguration.DEFAULT_LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS;
 import static org.apache.hudi.common.config.LockConfiguration.DEFAULT_LOCK_ACQUIRE_NUM_RETRIES;
 import static org.apache.hudi.common.config.LockConfiguration.DEFAULT_LOCK_ACQUIRE_WAIT_TIMEOUT_MS;
 import static org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_CLIENT_NUM_RETRIES_PROP_KEY;
@@ -76,10 +75,9 @@ public TimeGeneratorBase(HoodieTimeGeneratorConfig config, StorageConfiguration<
         Integer.parseInt(DEFAULT_LOCK_ACQUIRE_NUM_RETRIES));
     lockAcquireWaitTimeInMs = lockConfiguration.getConfig().getInteger(LOCK_ACQUIRE_WAIT_TIMEOUT_MS_PROP_KEY,
         DEFAULT_LOCK_ACQUIRE_WAIT_TIMEOUT_MS);
-    // The maximum time to wait for each time generation to resolve the clock skew issue on distributed hosts.
-    long maxWaitTimeInMs = lockConfiguration.getConfig().getLong(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY,
-        Long.parseLong(DEFAULT_LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS));
-    lockRetryHelper = new RetryHelper<>(maxWaitTimeInMs, maxRetries, maxWaitTimeInMs,
+    // The initial time to wait for each time generation to resolve the clock skew issue on distributed hosts.
+    long initialRetryInternal = lockConfiguration.getConfig().getLong(LOCK_ACQUIRE_CLIENT_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY, 200);
+    lockRetryHelper = new RetryHelper<>(initialRetryInternal, maxRetries, initialRetryInternal,
         Arrays.asList(HoodieLockException.class, InterruptedException.class), "acquire timeGenerator lock");
   }
 

hudi-common/src/main/java/org/apache/hudi/common/table/timeline/dto/InstantStateDTO.java

@@ -350,6 +350,7 @@ private boolean flushBucket(RowDataBucket bucket) {
     final List<WriteStatus> writeStatus = writeRecords(instant, bucket);
     final WriteMetadataEvent event = WriteMetadataEvent.builder()
         .taskID(taskID)
+        .checkpointId(this.checkpointId)
         .instantTime(instant) // the write instant may shift but the event still use the currentInstant.
         .writeStatus(writeStatus)
         .lastBatch(false)
@@ -386,6 +387,7 @@ private void flushRemaining(boolean endInput) {
     }
     final WriteMetadataEvent event = WriteMetadataEvent.builder()
         .taskID(taskID)
+        .checkpointId(checkpointId)
         .instantTime(currentInstant)
         .writeStatus(writeStatus)
         .lastBatch(true)
@@ -397,8 +399,6 @@ private void flushRemaining(boolean endInput) {
     this.tracer.reset();
     this.writeClient.cleanHandles();
     this.writeStatuses.addAll(writeStatus);
-    // blocks flushing until the coordinator starts a new instant
-    this.confirming = true;
 
     writeMetrics.endDataFlush();
     writeMetrics.resetAfterCommit();