Return on LA queue thread interrupts

Author: Sushisource

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

@@ -107,6 +107,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 +121,11 @@ private void processQueue() {
               LocalActivityResult.processingFailed(
                   executionContext.getActivityId(), request.task.getAttemptTask().getAttempt(), e));
         }
+        if (e.getCause() instanceof InterruptedException) {
+          // It's possible the interrupt happens inside the callback, so check that as well.
+          Thread.currentThread().interrupt();
+          return;
+        }
       }
     }
   }
@@ -170,11 +176,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 +194,7 @@ public void awaitTermination(long timeout, TimeUnit unit) {
       // timeout duration if no task was ever submitted.
       return;
     }
+
     ShutdownManager.awaitTermination(queueThreadService, unit.toMillis(timeout));
   }
 }