Fix shutting down status and add shutdown integration test

Author: Sushisource

.github/workflows/ci.yml

@@ -110,6 +110,7 @@ jobs:
               --dynamic-config-value history.enableRequestIdRefLinks=true \
               --dynamic-config-value frontend.WorkerHeartbeatsEnabled=true \
               --dynamic-config-value frontend.ListWorkersEnabled=true \
+              --dynamic-config-value frontend.enableCancelWorkerPollsOnShutdown=true \
               --dynamic-config-value 'component.callbacks.allowedAddresses=[{"Pattern":"localhost:7243","AllowInsecure":true}]' \
               --dynamic-config-value frontend.activityAPIsEnabled=true \
               --dynamic-config-value activity.enableStandalone=true \

temporal-sdk/src/main/java/io/temporal/worker/Worker.java

@@ -482,6 +482,7 @@ void start() {
   }
 
   CompletableFuture<Void> shutdown(ShutdownManager shutdownManager, boolean interruptUserTasks) {
+    shuttingDown = true;
     ShutdownWorkerRequest.Builder requestBuilder =
         ShutdownWorkerRequest.newBuilder()
             .setNamespace(namespace)

temporal-sdk/src/test/java/io/temporal/worker/WorkerShutdownTest.java

@@ -13,6 +13,8 @@
 import io.temporal.activity.ActivityInterface;
 import io.temporal.activity.ActivityMethod;
 import io.temporal.api.enums.v1.TaskQueueType;
+import io.temporal.api.enums.v1.WorkerStatus;
+import io.temporal.api.worker.v1.WorkerHeartbeat;
 import io.temporal.api.workflowservice.v1.GetSystemInfoResponse;
 import io.temporal.api.workflowservice.v1.ShutdownWorkerRequest;
 import io.temporal.api.workflowservice.v1.ShutdownWorkerResponse;
@@ -31,6 +33,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 import org.junit.Test;
 import org.mockito.ArgumentCaptor;
 
@@ -121,6 +124,9 @@ public void activeTaskQueueTypesEvaluatedAtShutdownTime() throws Exception {
     worker.registerWorkflowImplementationTypes(TestWorkflowImpl.class);
     worker.registerActivitiesImplementations(new TestActivityImpl());
     worker.registerNexusServiceImplementation(new TestNexusServiceImpl());
+    Supplier<WorkerHeartbeat> heartbeatSupplier =
+        () -> WorkerHeartbeat.newBuilder().setStatus(WorkerStatus.WORKER_STATUS_RUNNING).build();
+    worker.setHeartbeatSupplier(heartbeatSupplier);
 
     worker.shutdown(new ShutdownManager(), true).get(5, TimeUnit.SECONDS);
 
@@ -137,5 +143,9 @@ public void activeTaskQueueTypesEvaluatedAtShutdownTime() throws Exception {
     assertTrue(
         "ShutdownWorkerRequest should include NEXUS type registered after construction",
         shutdownTypes.contains(TaskQueueType.TASK_QUEUE_TYPE_NEXUS));
+    assertEquals(
+        "ShutdownWorkerRequest heartbeat should report SHUTTING_DOWN",
+        WorkerStatus.WORKER_STATUS_SHUTTING_DOWN,
+        captor.getValue().getWorkerHeartbeat().getStatus());
   }
 }