Add exception guards for scheduleWithFixedDelay

core/trino-main/src/main/java/io/trino/dispatcher/QueuedStatementResource.java

@@ -504,7 +504,15 @@ public QueryManager(Duration querySubmissionTimeout)
 
         public void initialize(DispatchManager dispatchManager)
         {
-            scheduledExecutorService.scheduleWithFixedDelay(() -> syncWith(dispatchManager), 200, 200, MILLISECONDS);
+            scheduledExecutorService.scheduleWithFixedDelay(() -> {
+                try {
+                    syncWith(dispatchManager);
+                }
+                catch (Throwable e) {
+                    // ignore to avoid getting unscheduled
+                    log.error(e, "Unexpected error synchronizing with dispatch manager");
+                }
+            }, 200, 200, MILLISECONDS);
         }
 
         public void destroy()

core/trino-main/src/main/java/io/trino/execution/scheduler/BinPackingNodeAllocatorService.java

@@ -165,7 +165,15 @@ public void start()
         }
 
         refreshNodePoolMemoryInfos();
-        executor.scheduleWithFixedDelay(this::refreshNodePoolMemoryInfos, 1, 1, TimeUnit.SECONDS);
+        executor.scheduleWithFixedDelay(() -> {
+            try {
+                refreshNodePoolMemoryInfos();
+            }
+            catch (Throwable e) {
+                // ignore to avoid getting unscheduled
+                log.error(e, "Unexpected error while refreshing node pool memory infos");
+            }
+        }, 1, 1, TimeUnit.SECONDS);
     }
 
     @PreDestroy

core/trino-main/src/main/java/io/trino/server/remotetask/TaskInfoFetcher.java

@@ -21,6 +21,7 @@
 import io.airlift.http.client.HttpUriBuilder;
 import io.airlift.http.client.Request;
 import io.airlift.json.JsonCodec;
+import io.airlift.log.Logger;
 import io.airlift.units.DataSize;
 import io.airlift.units.Duration;
 import io.opentelemetry.api.trace.SpanBuilder;
@@ -57,6 +58,8 @@
 
 public class TaskInfoFetcher
 {
+    private static final Logger log = Logger.get(TaskInfoFetcher.class);
+
     private final TaskId taskId;
     private final Consumer<Throwable> onFail;
     private final ContinuousTaskStatusFetcher taskStatusFetcher;
@@ -186,14 +189,20 @@ public SpoolingOutputStats.Snapshot retrieveAndDropSpoolingOutputStats()
     private synchronized void scheduleUpdate()
     {
         scheduledFuture = updateScheduledExecutor.scheduleWithFixedDelay(() -> {
-            synchronized (this) {
-                // if the previous request still running, don't schedule a new request
-                if (future != null && !future.isDone()) {
-                    return;
+            try {
+                synchronized (this) {
+                    // if the previous request still running, don't schedule a new request
+                    if (future != null && !future.isDone()) {
+                        return;
+                    }
+                }
+                if (nanosSince(lastUpdateNanos.get()).toMillis() >= updateIntervalMillis) {
+                    sendNextRequest();
                 }
             }
-            if (nanosSince(lastUpdateNanos.get()).toMillis() >= updateIntervalMillis) {
-                sendNextRequest();
+            catch (Throwable e) {
+                // ignore to avoid getting unscheduled
+                log.error(e, "Unexpected error while getting task info");
             }
         }, 0, 100, MILLISECONDS);
     }