Use long poll token in describeActivity.

Author: GregoryTravis

temporal-sdk/src/main/java/io/temporal/client/ActivityExecutionDescription.java

@@ -30,9 +30,13 @@ public final class ActivityExecutionDescription extends ActivityExecutionMetadat
   private final ActivityExecutionInfo info;
   private final DataConverter dataConverter;
   private final String namespace;
+  private final @Nullable byte[] longPollToken;
 
   public ActivityExecutionDescription(
-      ActivityExecutionInfo info, DataConverter dataConverter, String namespace) {
+      ActivityExecutionInfo info,
+      DataConverter dataConverter,
+      String namespace,
+      @Nullable byte[] longPollToken) {
     super(
         null,
         info.getActivityId(),
@@ -51,6 +55,7 @@ public ActivityExecutionDescription(
     this.info = info;
     this.dataConverter = dataConverter;
     this.namespace = namespace;
+    this.longPollToken = longPollToken;
   }
 
   private static @Nullable String nullIfEmpty(String s) {
@@ -131,6 +136,15 @@ public String getLastWorkerIdentity() {
     return w.isEmpty() ? null : w;
   }
 
+  /**
+   * Token for a follow-on {@link UntypedActivityHandle#describe(byte[])} call. Pass this token to
+   * long-poll until the activity state changes. {@code null} when the activity is complete.
+   */
+  @Nullable
+  public byte[] getLongPollToken() {
+    return longPollToken;
+  }
+
   /** Time when the next retry attempt will be scheduled. */
   @Nullable
   public Instant getNextAttemptScheduleTime() {

temporal-sdk/src/main/java/io/temporal/client/ActivityHandleWrapper.java

@@ -68,6 +68,11 @@ public ActivityExecutionDescription describe() {
     return delegate.describe();
   }
 
+  @Override
+  public ActivityExecutionDescription describe(@Nullable byte[] longPollToken) {
+    return delegate.describe(longPollToken);
+  }
+
   @Override
   public void cancel() {
     delegate.cancel();

temporal-sdk/src/main/java/io/temporal/client/UntypedActivityHandle.java

@@ -71,6 +71,18 @@ public interface UntypedActivityHandle {
    */
   ActivityExecutionDescription describe();
 
+  /**
+   * Long-polls until the activity state changes from the state encoded in {@code longPollToken},
+   * then returns the updated description. Pass the token from a previous {@link #describe()} call
+   * via {@link ActivityExecutionDescription#getLongPollToken()}. If {@code longPollToken} is {@code
+   * null}, returns the current state immediately (equivalent to {@link #describe()}).
+   *
+   * @param longPollToken token from a previous describe response, or {@code null} for an immediate
+   *     snapshot
+   * @return updated description of the activity
+   */
+  ActivityExecutionDescription describe(@Nullable byte[] longPollToken);
+
   /**
    * Requests cancellation of the activity. The activity will receive a cancellation via {@link
    * io.temporal.activity.ActivityExecutionContext#heartbeat(Object)}.

temporal-sdk/src/main/java/io/temporal/common/interceptors/ActivityClientCallsInterceptor.java

@@ -157,10 +157,17 @@ public R getResult() {
   final class DescribeActivityInput {
     private final String id;
     private final @Nullable String runId;
+    private final @Nullable byte[] longPollToken;
 
     public DescribeActivityInput(String id, @Nullable String runId) {
+      this(id, runId, null);
+    }
+
+    public DescribeActivityInput(
+        String id, @Nullable String runId, @Nullable byte[] longPollToken) {
       this.id = id;
       this.runId = runId;
+      this.longPollToken = longPollToken;
     }
 
     public String getId() {
@@ -171,6 +178,11 @@ public String getId() {
     public String getRunId() {
       return runId;
     }
+
+    @Nullable
+    public byte[] getLongPollToken() {
+      return longPollToken;
+    }
   }
 
   @Experimental

temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHandleImpl.java

@@ -64,9 +64,15 @@ public <R> CompletableFuture<R> getResultAsync(Class<R> resultClass, @Nullable T
 
   @Override
   public ActivityExecutionDescription describe() {
+    return describe(null);
+  }
+
+  @Override
+  public ActivityExecutionDescription describe(@Nullable byte[] longPollToken) {
     return clientCallsInterceptor
         .describeActivity(
-            new ActivityClientCallsInterceptor.DescribeActivityInput(activityId, activityRunId))
+            new ActivityClientCallsInterceptor.DescribeActivityInput(
+                activityId, activityRunId, longPollToken))
         .getDescription();
   }
 

temporal-sdk/src/main/java/io/temporal/internal/client/RootActivityClientInvoker.java

@@ -186,10 +186,18 @@ public DescribeActivityOutput describeActivity(DescribeActivityInput input) {
     if (input.getRunId() != null) {
       req.setRunId(input.getRunId());
     }
+    if (input.getLongPollToken() != null) {
+      req.setLongPollToken(ByteString.copyFrom(input.getLongPollToken()));
+    }
     DescribeActivityExecutionResponse response = genericClient.describeActivity(req.build());
+    byte[] token =
+        response.getLongPollToken().isEmpty() ? null : response.getLongPollToken().toByteArray();
     return new DescribeActivityOutput(
         new ActivityExecutionDescription(
-            response.getInfo(), clientOptions.getDataConverter(), clientOptions.getNamespace()));
+            response.getInfo(),
+            clientOptions.getDataConverter(),
+            clientOptions.getNamespace(),
+            token));
   }
 
   @Override

temporal-sdk/src/test/java/io/temporal/client/ActivityExecutionDescriptionTest.java

@@ -38,14 +38,14 @@ private ActivityExecutionInfo buildInfo(String activityId, String runId) {
   @Test
   public void testNullRunIdWhenEmpty() {
     ActivityExecutionDescription desc =
-        new ActivityExecutionDescription(buildInfo("act-id", ""), CONVERTER, "test-ns");
+        new ActivityExecutionDescription(buildInfo("act-id", ""), CONVERTER, "test-ns", null);
     assertNull(desc.getActivityRunId());
   }
 
   @Test
   public void testNullableFieldsAbsentByDefault() {
     ActivityExecutionDescription desc =
-        new ActivityExecutionDescription(buildInfo("act-id", ""), CONVERTER, "test-ns");
+        new ActivityExecutionDescription(buildInfo("act-id", ""), CONVERTER, "test-ns", null);
 
     assertNull(desc.getCloseTime());
     assertNull(desc.getExecutionDuration());
@@ -69,14 +69,14 @@ public void testNullableFieldsAbsentByDefault() {
   @Test
   public void testScheduledTime() {
     ActivityExecutionDescription desc =
-        new ActivityExecutionDescription(buildInfo("act-id", ""), CONVERTER, "test-ns");
+        new ActivityExecutionDescription(buildInfo("act-id", ""), CONVERTER, "test-ns", null);
     assertEquals(Instant.ofEpochMilli(1000), desc.getScheduledTime());
   }
 
   @Test
   public void testHasHeartbeatDetailsAbsent() {
     ActivityExecutionDescription desc =
-        new ActivityExecutionDescription(buildInfo("id", "run"), CONVERTER, "test-ns");
+        new ActivityExecutionDescription(buildInfo("id", "run"), CONVERTER, "test-ns", null);
     assertFalse(desc.hasHeartbeatDetails());
     assertFalse(desc.getHeartbeatDetails(String.class).isPresent());
   }
@@ -87,7 +87,7 @@ public void testGetHeartbeatDetailsPresent() {
     ActivityExecutionInfo info =
         buildInfo("id", "run").toBuilder().setHeartbeatDetails(encoded).build();
     ActivityExecutionDescription desc =
-        new ActivityExecutionDescription(info, CONVERTER, "test-ns");
+        new ActivityExecutionDescription(info, CONVERTER, "test-ns", null);
 
     assertTrue(desc.hasHeartbeatDetails());
     Optional<String> result = desc.getHeartbeatDetails(String.class);
@@ -103,7 +103,7 @@ public void testGetHeartbeatDetailsWithExplicitGenericType() {
     ActivityExecutionInfo info =
         buildInfo("id", "run").toBuilder().setHeartbeatDetails(encoded).build();
     ActivityExecutionDescription desc =
-        new ActivityExecutionDescription(info, CONVERTER, "test-ns");
+        new ActivityExecutionDescription(info, CONVERTER, "test-ns", null);
 
     Type genericType = new TypeToken<List<String>>() {}.getType();
     Class<List<String>> listClass = (Class<List<String>>) (Class<?>) List.class;
@@ -122,7 +122,7 @@ public void testGetWorkerDeploymentVersionPresent() {
     ActivityExecutionInfo info =
         buildInfo("id", "run").toBuilder().setLastDeploymentVersion(protoVersion).build();
     ActivityExecutionDescription desc =
-        new ActivityExecutionDescription(info, CONVERTER, "test-ns");
+        new ActivityExecutionDescription(info, CONVERTER, "test-ns", null);
 
     WorkerDeploymentVersion version = desc.getWorkerDeploymentVersion();
     assertNotNull(version);
@@ -137,7 +137,7 @@ public void testGetPriorityPresent() {
     ActivityExecutionInfo info =
         buildInfo("id", "run").toBuilder().setPriority(protoPriority).build();
     ActivityExecutionDescription desc =
-        new ActivityExecutionDescription(info, CONVERTER, "test-ns");
+        new ActivityExecutionDescription(info, CONVERTER, "test-ns", null);
 
     Priority priority = desc.getPriority();
     assertNotNull(priority);

temporal-sdk/src/test/java/io/temporal/client/functional/StandaloneActivityTest.java

@@ -392,6 +392,43 @@ public void testDescribeUserMetadataIsAccurate() {
     assertEquals("Test details\nLine 2", desc.getStaticDetails());
   }
 
+  @Test
+  public void testDescribeLongPollObservesCompletion() throws Exception {
+    assumeTrue(SDKTestWorkflowRule.useExternalService);
+    asyncStartLatch = new CountDownLatch(1);
+    try {
+      ActivityClient client = newActivityClient();
+      ActivityHandle<String> handle =
+          client.start(
+              AsyncCompletionActivity.class,
+              AsyncCompletionActivity::complete,
+              simpleOpts(uniqueId()));
+
+      assertTrue("Activity did not start within 30s", asyncStartLatch.await(30, TimeUnit.SECONDS));
+
+      ActivityExecutionDescription desc = handle.describe();
+      assertEquals(ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_RUNNING, desc.getStatus());
+      assertNotNull("Running activity must have a long-poll token", desc.getLongPollToken());
+
+      // Long-poll in background — blocks until server detects a state change
+      CompletableFuture<ActivityExecutionDescription> longPollFuture =
+          CompletableFuture.supplyAsync(() -> handle.describe(desc.getLongPollToken()));
+
+      // Complete the activity externally so the server unblocks the poll
+      client
+          .newActivityCompletionClient()
+          .completeStandalone(asyncActivityId, Optional.empty(), "long-poll-result");
+
+      ActivityExecutionDescription updated = longPollFuture.get(30, TimeUnit.SECONDS);
+      assertEquals(
+          ActivityExecutionStatus.ACTIVITY_EXECUTION_STATUS_COMPLETED, updated.getStatus());
+    } finally {
+      asyncStartLatch = null;
+      asyncActivityId = null;
+      asyncActivityRunId = null;
+    }
+  }
+
   @Test
   public void testCancelRunningActivitySucceeds() throws InterruptedException {
     assumeTrue(SDKTestWorkflowRule.useExternalService);