Return on LA queue thread interrupts

Author: Sushisource

temporal-sdk/src/main/java/io/temporal/internal/worker/LocalActivitySlotSupplierQueue.java

@@ -83,7 +83,11 @@ private void processQueue() {
       SlotPermit slotPermit = null;
       QueuedLARequest request = null;
       try {
-        request = requestQueue.take();
+        // Use poll with timeout instead of take() to avoid being blocked indefinitely
+        request = requestQueue.poll(100, TimeUnit.MILLISECONDS);
+        if (request == null) {
+          continue;
+        }
 
         SlotSupplierFuture future = slotSupplier.reserveSlot(request.data);
         try {
@@ -107,6 +111,7 @@ private void processQueue() {
         afterReservedCallback.apply(request.task);
       } catch (InterruptedException e) {
         Thread.currentThread().interrupt();
+        return;
       } catch (Throwable e) {
         // Fail the workflow task if something went wrong executing the local activity (at the
         // executor level, otherwise, the LA handler itself should be handling errors)
@@ -120,6 +125,10 @@ private void processQueue() {
               LocalActivityResult.processingFailed(
                   executionContext.getActivityId(), request.task.getAttemptTask().getAttempt(), e));
         }
+        if (e.getCause() instanceof InterruptedException) {
+          Thread.currentThread().interrupt();
+          return;
+        }
       }
     }
   }
@@ -170,11 +179,9 @@ public boolean isTerminated() {
   @Override
   public CompletableFuture<Void> shutdown(ShutdownManager shutdownManager, boolean interruptTasks) {
     running = false;
-    if (requestQueue.isEmpty()) {
-      // Just interrupt the thread, so that if we're waiting on blocking take the thread will
-      // be interrupted and exit. Otherwise the loop will exit once the queue is empty.
-      queueThreadService.shutdownNow();
-    }
+    // Always interrupt. This won't cause any *tasks* to be interrupted, since the queue thread is
+    // only responsible for handing them out.
+    queueThreadService.shutdownNow();
 
     return interruptTasks
         ? shutdownManager.shutdownExecutorNowUntimed(
@@ -190,6 +197,7 @@ public void awaitTermination(long timeout, TimeUnit unit) {
       // timeout duration if no task was ever submitted.
       return;
     }
+
     ShutdownManager.awaitTermination(queueThreadService, unit.toMillis(timeout));
   }
 }