[hudi] branch master updated: [HUDI-4574] Fixed timeline based marker thread safety issue (#6383)

[danny0405]

This is an automated email from the ASF dual-hosted git repository.

danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git


The following commit(s) were added to refs/heads/master by this push:
     new bad954c3b2 [HUDI-4574] Fixed timeline based marker thread safety issue 
(#6383)
bad954c3b2 is described below

commit bad954c3b27b9e5236b93c6b0e2be219337fa179
Author: novisfff <62633257+novis...@users.noreply.github.com>
AuthorDate: Tue Aug 16 10:40:09 2022 +0800

    [HUDI-4574] Fixed timeline based marker thread safety issue (#6383)
    
    * fixed timeline based markers thread safety issue
    * add document for TimelineBasedMarkers thread safety issues
---
 .../org/apache/hudi/timeline/service/handlers/MarkerHandler.java    | 5 +++--
 .../service/handlers/marker/MarkerCreationDispatchingRunnable.java  | 6 ++++--
 2 files changed, 7 insertions(+), 4 deletions(-)

diff --git 
a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java
 
b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java
index e793c20432..32b85bdfc3 100644
--- 
a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java
+++ 
b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/MarkerHandler.java
@@ -34,7 +34,7 @@ import org.apache.log4j.LogManager;
 import org.apache.log4j.Logger;
 
 import java.io.IOException;
-import java.util.HashMap;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
@@ -74,7 +74,8 @@ public class MarkerHandler extends Handler {
   // Parallelism for reading and deleting marker files
   private final int parallelism;
   // Marker directory states, {markerDirPath -> MarkerDirState instance}
-  private final Map<String, MarkerDirState> markerDirStateMap = new 
HashMap<>();
+  // Use ConcurrentHashMap to ensure thread safety in 
dispatchingExecutorService
+  private final Map<String, MarkerDirState> markerDirStateMap = new 
ConcurrentHashMap<>();
   // A thread to dispatch marker creation requests to batch processing threads
   private final MarkerCreationDispatchingRunnable 
markerCreationDispatchingRunnable;
   private final Object firstCreationRequestSeenLock = new Object();
diff --git 
a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationDispatchingRunnable.java
 
b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationDispatchingRunnable.java
index 93093a4b4c..5f647e9b3f 100644
--- 
a/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationDispatchingRunnable.java
+++ 
b/hudi-timeline-service/src/main/java/org/apache/hudi/timeline/service/handlers/marker/MarkerCreationDispatchingRunnable.java
@@ -66,8 +66,10 @@ public class MarkerCreationDispatchingRunnable implements 
Runnable {
 
     // Only fetch pending marker creation requests that can be processed,
     // i.e., that markers can be written to a underlying file
-    for (String markerDir : markerDirStateMap.keySet()) {
-      MarkerDirState markerDirState = markerDirStateMap.get(markerDir);
+    // markerDirStateMap is used in other thread, need to ensure thread safety
+    for (Map.Entry<String, MarkerDirState> entry : 
markerDirStateMap.entrySet()) {
+      String markerDir = entry.getKey();
+      MarkerDirState markerDirState = entry.getValue();
       Option<Integer> fileIndex = markerDirState.getNextFileIndexToUse();
       if (!fileIndex.isPresent()) {
         LOG.debug("All marker files are busy, skip batch processing of create 
marker requests in " + markerDir);