fix: detect shutdown and break DurableTaskGrpcWorker loop

The worker loop ran `while (true)` and only exited if an
InterruptedException happened during the 5-second retry sleep.
If `close()` was called while the gRPC stream was blocking, the
CANCELLED exception was logged but the loop kept retrying.

- Replace `while (true)` with a check on `isNormalShutdown` and
  the thread interrupt flag so the loop exits promptly.
- Break out of the retry path on CANCELLED when `isNormalShutdown`
  is set, avoiding a misleading 5-second sleep after `close()`.
- Re-set the interrupt flag before breaking on InterruptedException
  to preserve the interrupt contract for callers higher up.

Signed-off-by: Javier Aliaga <javier@diagrid.io>

Author: javier-aliaga

durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java

@@ -171,7 +171,7 @@ public void startAndBlock() {
         this.dataConverter,
         logger);
 
-    while (true) {
+    while (!this.isNormalShutdown && !Thread.currentThread().isInterrupted()) {
       try {
         OrchestratorService.GetWorkItemsRequest getWorkItemsRequest = OrchestratorService.GetWorkItemsRequest
             .newBuilder().build();
@@ -210,6 +210,9 @@ public void startAndBlock() {
               this.getSidecarAddress());
         } else if (e.getStatus().getCode() == Status.Code.CANCELLED) {
           logger.log(Level.INFO, "Durable Task worker has disconnected from {0}.", this.getSidecarAddress());
+          if (this.isNormalShutdown) {
+            break;
+          }
         } else {
           logger.log(Level.WARNING,
               String.format("Unexpected failure connecting to %s", this.getSidecarAddress()), e);
@@ -219,6 +222,7 @@ public void startAndBlock() {
         try {
           Thread.sleep(5000);
         } catch (InterruptedException ex) {
+          Thread.currentThread().interrupt();
           break;
         }
       }