Conditional waiting on the previous polling task

application/src/main/java/org/opentripplanner/updater/spi/PollingGraphUpdater.java

@@ -3,6 +3,7 @@
 import java.time.Duration;
 import java.util.concurrent.CancellationException;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import org.opentripplanner.framework.application.OTPFeature;
 import org.opentripplanner.updater.GraphWriterRunnable;
 import org.slf4j.Logger;
@@ -42,6 +43,12 @@ public abstract class PollingGraphUpdater implements GraphUpdater {
    */
   protected WriteToGraphCallback saveResultOnGraph;
 
+  /**
+   * Handle on the previous polling execution.
+   * Initially null when the polling updater starts.
+   */
+  private Future<?> previousTask;
+
   /** Shared configuration code for all polling graph updaters. */
   protected PollingGraphUpdater(PollingGraphUpdaterParameters config) {
     this.pollingPeriod = config.frequency();
@@ -55,6 +62,10 @@ public Duration pollingPeriod() {
   @Override
   public final void run() {
     try {
+      if (OTPFeature.WaitForGraphUpdateInPollingUpdaters.isOn()) {
+        waitForPreviousTask();
+      }
+
       // Run concrete polling graph updater's implementation method.
       runPolling();
       if (runOnlyOnce()) {
@@ -113,11 +124,28 @@ public final void setup(WriteToGraphCallback writeToGraphCallback) {
    */
   protected abstract void runPolling() throws Exception;
 
-  protected final void updateGraph(GraphWriterRunnable task)
-    throws ExecutionException, InterruptedException {
-    var result = saveResultOnGraph.execute(task);
-    if (OTPFeature.WaitForGraphUpdateInPollingUpdaters.isOn()) {
-      result.get();
+  /**
+   * Post the update task in the GraphWriter queue.
+   * This is non-blocking.
+   */
+  protected final void updateGraph(GraphWriterRunnable task) {
+    previousTask = saveResultOnGraph.execute(task);
+  }
+
+  /**
+   * If the previous task takes longer than the polling interval,
+   * we delay the next polling cycle until the task is complete.
+   * This prevents tasks from piling up.
+   * */
+  private void waitForPreviousTask() throws InterruptedException, ExecutionException {
+    if (previousTask != null && !previousTask.isDone()) {
+      LOG.info("Delaying polling until the previous task is complete");
+      long startBlockingWait = System.currentTimeMillis();
+      previousTask.get();
+      LOG.info(
+        "Resuming polling after waiting an additional {}s",
+        (System.currentTimeMillis() - startBlockingWait) / 1000
+      );
     }
   }
 }