[GitHub] [hudi] zhangyue19921010 commented on a diff in pull request #7582: [HUDI-5488]Make sure Discrupt queue start first, then insert records

[GitBox]

zhangyue19921010 commented on code in PR #7582:
URL: https://github.com/apache/hudi/pull/7582#discussion_r1059638481


##########
hudi-common/src/main/java/org/apache/hudi/common/util/queue/DisruptorMessageQueue.java:
##########
@@ -105,6 +113,12 @@ protected void setHandlers(HoodieConsumer consumer) {
 
   protected void start() {
     queue.start();
+    synchronized (this) {
+      if (!isStart) {
+        isStart = true;
+        notifyAll();
+      }
+    }

Review Comment:
   ```suggestion
       synchronized (this) {
         if (!isStart) {
           queue.start();
           isStart = true;
           notifyAll();
         }
       }
   ```



##########
hudi-common/src/main/java/org/apache/hudi/common/util/queue/DisruptorMessageQueue.java:
##########
@@ -60,6 +61,12 @@ public long size() {
 
   @Override
   public void insertRecord(I value) throws Exception {
+    if (!isStart) {
+      synchronized (this) {
+        wait();
+      }
+    }

Review Comment:
   ```suggestion
         synchronized (this) {
             If (!isStart) {
                  wait();
           }
       }
   ```
   
   We need to let synchronized cover isStart check in case that `isStart ` is 
changed after `if (!isStart)` but before `wait();`
   
   But in this way, we will do synchronized check for each record ingestion 
right?
   
   



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscr...@hudi.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org