Extract method HttpRemoteTask#triggerUpdate

trinodb · Jul 9, 2021 · 613bd2f · 613bd2f
1 parent 641aaac
commit 613bd2f
Showing 1 changed file with 11 additions and 8 deletions.
diff --git a/core/trino-main/src/main/java/io/trino/server/remotetask/HttpRemoteTask.java b/core/trino-main/src/main/java/io/trino/server/remotetask/HttpRemoteTask.java
@@ -359,8 +359,7 @@ public synchronized void addSplits(Multimap<PlanNodeId, Split> splitsBySource)
         updateSplitQueueSpace();
 
         if (needsUpdate) {
-            this.needsUpdate.set(true);
-            scheduleUpdate();
+            triggerUpdate();
         }
     }
 
@@ -372,16 +371,14 @@ public synchronized void noMoreSplits(PlanNodeId sourceId)
         }
 
         noMoreSplits.put(sourceId, true);
-        needsUpdate.set(true);
-        scheduleUpdate();
+        triggerUpdate();
     }
 
     @Override
     public synchronized void noMoreSplits(PlanNodeId sourceId, Lifespan lifespan)
     {
         if (pendingNoMoreSplitsForLifespan.put(sourceId, lifespan)) {
-            needsUpdate.set(true);
-            scheduleUpdate();
+            triggerUpdate();
         }
     }
 
@@ -394,8 +391,7 @@ public synchronized void setOutputBuffers(OutputBuffers newOutputBuffers)
 
         if (newOutputBuffers.getVersion() > outputBuffers.get().getVersion()) {
             outputBuffers.set(newOutputBuffers);
-            needsUpdate.set(true);
-            scheduleUpdate();
+            triggerUpdate();
         }
     }
 
@@ -511,6 +507,13 @@ private void scheduleUpdate()
         executor.execute(this::sendUpdate);
     }
 
+    private synchronized void triggerUpdate()
+    {
+        // synchronized so that needsUpdate is not cleared in sendUpdate before actual request is sent
+        needsUpdate.set(true);
+        sendUpdate();
+    }
+
     private synchronized void sendUpdate()
     {
         TaskStatus taskStatus = getTaskStatus();