From b867534a1f5c1bbf358d09321de7f14554017332 Mon Sep 17 00:00:00 2001 From: Mana Agrawal Date: Thu, 6 Aug 2026 14:44:52 -0700 Subject: [PATCH 1/4] Fix ManualActivityCompletionClient.recordHeartbeat: swallowed exceptions and missing retry Two related bugs in the same method: Fixes #2983: recordHeartbeat wrapped the RPC call and the response-flag checks in a single try block, so its own ActivityCanceledException / ActivityResetException / ActivityPausedException were caught by the generic catch (Exception e) and turned into ActivityCompletionFailureException by processException. Callers could not tell a cancelled/reset/paused activity apart from a failed RPC without unwrapping getCause(). The interface also declared `throws CanceledFailure`, a type the method has never actually thrown (CanceledFailure and ActivityCompletionException are unrelated siblings under TemporalException) -- changed to the type it genuinely throws, ActivityCompletionException. Fixes #2984: recordHeartbeat was the only one of this class's four RPC methods (complete/fail/reportCancellation/recordHeartbeat) that didn't go through grpcRetryer.retryWithResult(...). A single transient error (RESOURCE_EXHAUSTED from namespace rate limiting, DEADLINE_EXCEEDED, UNAVAILABLE) could fail the heartbeat outright, which for async completion can cost a long-running activity via heartbeat timeout. Now wrapped in the same retry helper the sibling methods already use. Both fixes land in the same restructure: the RPC call is now isolated in its own try/catch (wrapped in retryWithResult), and the cancel/reset/paused flag checks happen outside that catch so the correct exception always reaches the caller. --- .../ManualActivityCompletionClient.java | 10 +- .../ManualActivityCompletionClientImpl.java | 92 +++++++----- ...anualActivityCompletionClientImplTest.java | 134 ++++++++++++++++++ 3 files changed, 196 insertions(+), 40 deletions(-) create mode 100644 temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java diff --git a/temporal-sdk/src/main/java/io/temporal/activity/ManualActivityCompletionClient.java b/temporal-sdk/src/main/java/io/temporal/activity/ManualActivityCompletionClient.java index c1d7878156..34f30745ff 100644 --- a/temporal-sdk/src/main/java/io/temporal/activity/ManualActivityCompletionClient.java +++ b/temporal-sdk/src/main/java/io/temporal/activity/ManualActivityCompletionClient.java @@ -1,6 +1,6 @@ package io.temporal.activity; -import io.temporal.failure.CanceledFailure; +import io.temporal.client.ActivityCompletionException; import javax.annotation.Nonnull; import javax.annotation.Nullable; @@ -30,8 +30,14 @@ public interface ManualActivityCompletionClient { * Records heartbeat for an activity * * @param details to record with the heartbeat + * @throws ActivityCompletionException if the server reports the activity was cancelled, reset, or + * paused ({@link io.temporal.client.ActivityCanceledException}, {@link + * io.temporal.client.ActivityResetException}, {@link + * io.temporal.client.ActivityPausedException}), or if the heartbeat RPC itself fails after + * retries ({@link io.temporal.client.ActivityCompletionFailureException}, {@link + * io.temporal.client.ActivityNotExistsException}). */ - void recordHeartbeat(@Nullable Object details) throws CanceledFailure; + void recordHeartbeat(@Nullable Object details) throws ActivityCompletionException; /** * Confirms successful cancellation to the server. diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java index 0e68b107b5..0ae121fe14 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java @@ -13,7 +13,6 @@ import io.temporal.api.workflowservice.v1.*; import io.temporal.client.*; import io.temporal.common.converter.DataConverter; -import io.temporal.failure.CanceledFailure; import io.temporal.internal.client.ActivityClientHelper; import io.temporal.internal.common.OptionsUtils; import io.temporal.internal.retryer.GrpcRetryer; @@ -175,44 +174,61 @@ public void fail(@Nonnull Throwable exception) { } @Override - public void recordHeartbeat(@Nullable Object details) throws CanceledFailure { - try { - if (taskToken != null) { - RecordActivityTaskHeartbeatResponse status = - ActivityClientHelper.sendHeartbeatRequest( - service, - namespace, - identity, - taskToken, - dataConverterWithActivityExecutionContext.toPayloads(details), - metricsScope); - if (status.getCancelRequested()) { - throw new ActivityCanceledException(); - } else if (status.getActivityReset()) { - throw new ActivityResetException(); - } else if (status.getActivityPaused()) { - throw new ActivityPausedException(); - } - } else { - RecordActivityTaskHeartbeatByIdResponse status = - ActivityClientHelper.recordActivityTaskHeartbeatById( - service, - namespace, - identity, - execution, - activityId, - dataConverterWithActivityExecutionContext.toPayloads(details), - metricsScope); - if (status.getCancelRequested()) { - throw new ActivityCanceledException(); - } else if (status.getActivityReset()) { - throw new ActivityResetException(); - } else if (status.getActivityPaused()) { - throw new ActivityPausedException(); - } + public void recordHeartbeat(@Nullable Object details) throws ActivityCompletionException { + if (taskToken != null) { + RecordActivityTaskHeartbeatResponse status; + try { + status = + grpcRetryer.retryWithResult( + () -> + ActivityClientHelper.sendHeartbeatRequest( + service, + namespace, + identity, + taskToken, + dataConverterWithActivityExecutionContext.toPayloads(details), + metricsScope), + replyGrpcRetryerOptions); + } catch (Exception e) { + processException(e); + return; + } + if (status.getCancelRequested()) { + throw new ActivityCanceledException(); + } else if (status.getActivityReset()) { + throw new ActivityResetException(); + } else if (status.getActivityPaused()) { + throw new ActivityPausedException(); + } + } else { + if (activityId == null) { + throw new IllegalArgumentException("Either activity id or task token are required"); + } + RecordActivityTaskHeartbeatByIdResponse status; + try { + status = + grpcRetryer.retryWithResult( + () -> + ActivityClientHelper.recordActivityTaskHeartbeatById( + service, + namespace, + identity, + execution, + activityId, + dataConverterWithActivityExecutionContext.toPayloads(details), + metricsScope), + replyGrpcRetryerOptions); + } catch (Exception e) { + processException(e); + return; + } + if (status.getCancelRequested()) { + throw new ActivityCanceledException(); + } else if (status.getActivityReset()) { + throw new ActivityResetException(); + } else if (status.getActivityPaused()) { + throw new ActivityPausedException(); } - } catch (Exception e) { - processException(e); } } diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java new file mode 100644 index 0000000000..73eac7545a --- /dev/null +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java @@ -0,0 +1,134 @@ +package io.temporal.internal.client.external; + +import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.uber.m3.tally.NoopScope; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; +import io.temporal.api.common.v1.WorkflowExecution; +import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdResponse; +import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; +import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; +import io.temporal.client.ActivityCanceledException; +import io.temporal.client.ActivityCompletionFailureException; +import io.temporal.client.ActivityPausedException; +import io.temporal.client.ActivityResetException; +import io.temporal.common.converter.GlobalDataConverter; +import io.temporal.serviceclient.WorkflowServiceStubs; +import io.temporal.serviceclient.WorkflowServiceStubsOptions; +import org.junit.Before; +import org.junit.Test; + +public class ManualActivityCompletionClientImplTest { + + private WorkflowServiceStubs service; + private WorkflowServiceGrpc.WorkflowServiceBlockingStub blockingStub; + + @Before + public void setUp() { + service = mock(WorkflowServiceStubs.class); + blockingStub = mock(WorkflowServiceGrpc.WorkflowServiceBlockingStub.class); + when(service.blockingStub()).thenReturn(blockingStub); + when(blockingStub.withOption(any(), any())).thenReturn(blockingStub); + when(service.getServerCapabilities()) + .thenReturn( + () -> + io.temporal.api.workflowservice.v1.GetSystemInfoResponse.Capabilities + .getDefaultInstance()); + when(service.getOptions()) + .thenReturn(WorkflowServiceStubsOptions.newBuilder().validateAndBuildWithDefaults()); + } + + private ManualActivityCompletionClientImpl clientWithTaskToken() { + return new ManualActivityCompletionClientImpl( + service, + "test-namespace", + "test-identity", + GlobalDataConverter.get(), + new NoopScope(), + new byte[] {1, 2, 3}, + null, + null, + null); + } + + private ManualActivityCompletionClientImpl clientWithActivityId() { + return new ManualActivityCompletionClientImpl( + service, + "test-namespace", + "test-identity", + GlobalDataConverter.get(), + new NoopScope(), + null, + WorkflowExecution.newBuilder().setWorkflowId("wf").setRunId("run").build(), + "test-activity-id", + null); + } + + @Test + public void cancelRequestedThrowsActivityCanceledExceptionNotSwallowed() { + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenReturn( + RecordActivityTaskHeartbeatResponse.newBuilder().setCancelRequested(true).build()); + + assertThrows( + ActivityCanceledException.class, () -> clientWithTaskToken().recordHeartbeat("details")); + } + + @Test + public void activityResetThrowsActivityResetExceptionNotSwallowed() { + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenReturn( + RecordActivityTaskHeartbeatResponse.newBuilder().setActivityReset(true).build()); + + assertThrows( + ActivityResetException.class, () -> clientWithTaskToken().recordHeartbeat("details")); + } + + @Test + public void activityPausedThrowsActivityPausedExceptionNotSwallowed() { + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenReturn( + RecordActivityTaskHeartbeatResponse.newBuilder().setActivityPaused(true).build()); + + assertThrows( + ActivityPausedException.class, () -> clientWithTaskToken().recordHeartbeat("details")); + } + + @Test + public void byIdCancelRequestedThrowsActivityCanceledExceptionNotSwallowed() { + when(blockingStub.recordActivityTaskHeartbeatById(any())) + .thenReturn( + RecordActivityTaskHeartbeatByIdResponse.newBuilder().setCancelRequested(true).build()); + + assertThrows( + ActivityCanceledException.class, () -> clientWithActivityId().recordHeartbeat("details")); + } + + @Test + public void transientRpcErrorIsRetriedThenSucceeds() { + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow(new StatusRuntimeException(Status.RESOURCE_EXHAUSTED)) + .thenReturn(RecordActivityTaskHeartbeatResponse.getDefaultInstance()); + + // Should not throw: the transient error is retried and the second attempt succeeds. + clientWithTaskToken().recordHeartbeat("details"); + + verify(blockingStub, times(2)).recordActivityTaskHeartbeat(any()); + } + + @Test + public void nonTransientRpcErrorIsReportedAsActivityCompletionFailureException() { + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow(new StatusRuntimeException(Status.INTERNAL)); + + assertThrows( + ActivityCompletionFailureException.class, + () -> clientWithTaskToken().recordHeartbeat("details")); + } +} From f357900b4d9faa0e3144c221e3c93bc317dc84e3 Mon Sep 17 00:00:00 2001 From: Dan Plyukhin Date: Mon, 24 Aug 2026 16:51:36 -0500 Subject: [PATCH 2/4] Simplify ManualActivityCompletionClientImpl and add ActivityHeartbeatResponse container --- .../activity/HeartbeatContextImpl.java | 4 +- .../internal/client/ActivityClientHelper.java | 26 +++++---- .../client/ActivityHeartbeatResponse.java | 32 +++++++++++ .../ManualActivityCompletionClientImpl.java | 56 +++++++------------ 4 files changed, 70 insertions(+), 48 deletions(-) create mode 100644 temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHeartbeatResponse.java diff --git a/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java index 91da94ab0a..b471b6d74a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/activity/HeartbeatContextImpl.java @@ -7,12 +7,12 @@ import io.temporal.activity.ActivityInfo; import io.temporal.api.common.v1.Payloads; import io.temporal.api.enums.v1.TimeoutType; -import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; import io.temporal.client.*; import io.temporal.common.CancellationToken; import io.temporal.common.converter.DataConverter; import io.temporal.failure.TimeoutFailure; import io.temporal.internal.client.ActivityClientHelper; +import io.temporal.internal.client.ActivityHeartbeatResponse; import io.temporal.internal.concurrent.structured.CancelSource; import io.temporal.payload.context.ActivitySerializationContext; import io.temporal.serviceclient.WorkflowServiceStubs; @@ -332,7 +332,7 @@ private void checkHeartbeatTimeoutDeadlineLocked() { private void sendHeartbeatRequest(Object details) { try { - RecordActivityTaskHeartbeatResponse status = + ActivityHeartbeatResponse status = ActivityClientHelper.sendHeartbeatRequest( service, namespace, diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java index eb3e98107c..e272ec7ab9 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityClientHelper.java @@ -24,7 +24,7 @@ public final class ActivityClientHelper { private ActivityClientHelper() {} - public static RecordActivityTaskHeartbeatResponse sendHeartbeatRequest( + public static ActivityHeartbeatResponse sendHeartbeatRequest( WorkflowServiceStubs service, String namespace, String identity, @@ -37,13 +37,16 @@ public static RecordActivityTaskHeartbeatResponse sendHeartbeatRequest( .setNamespace(namespace) .setIdentity(identity); payloads.ifPresent(request::setDetails); - return service - .blockingStub() - .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) - .recordActivityTaskHeartbeat(request.build()); + RecordActivityTaskHeartbeatResponse response = + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .recordActivityTaskHeartbeat(request.build()); + return new ActivityHeartbeatResponse( + response.getCancelRequested(), response.getActivityReset(), response.getActivityPaused()); } - public static RecordActivityTaskHeartbeatByIdResponse recordActivityTaskHeartbeatById( + public static ActivityHeartbeatResponse recordActivityTaskHeartbeatById( WorkflowServiceStubs service, String namespace, String identity, @@ -60,9 +63,12 @@ public static RecordActivityTaskHeartbeatByIdResponse recordActivityTaskHeartbea .setNamespace(namespace) .setIdentity(identity); payloads.ifPresent(request::setDetails); - return service - .blockingStub() - .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) - .recordActivityTaskHeartbeatById(request.build()); + RecordActivityTaskHeartbeatByIdResponse response = + service + .blockingStub() + .withOption(METRICS_TAGS_CALL_OPTIONS_KEY, metricsScope) + .recordActivityTaskHeartbeatById(request.build()); + return new ActivityHeartbeatResponse( + response.getCancelRequested(), response.getActivityReset(), response.getActivityPaused()); } } diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHeartbeatResponse.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHeartbeatResponse.java new file mode 100644 index 0000000000..c17c1f77ee --- /dev/null +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHeartbeatResponse.java @@ -0,0 +1,32 @@ +package io.temporal.internal.client; + +import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdResponse; +import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; + +/** Container class to deduplicate {@link RecordActivityTaskHeartbeatByIdResponse} and + * {@link RecordActivityTaskHeartbeatResponse}. + */ +public final class ActivityHeartbeatResponse { + private final boolean cancelRequested; + private final boolean activityReset; + private final boolean activityPaused; + + ActivityHeartbeatResponse( + boolean cancelRequested, boolean activityReset, boolean activityPaused) { + this.cancelRequested = cancelRequested; + this.activityReset = activityReset; + this.activityPaused = activityPaused; + } + + public boolean getCancelRequested() { + return cancelRequested; + } + + public boolean getActivityReset() { + return activityReset; + } + + public boolean getActivityPaused() { + return activityPaused; + } +} diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java index 0ae121fe14..30738a1467 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java @@ -14,6 +14,7 @@ import io.temporal.client.*; import io.temporal.common.converter.DataConverter; import io.temporal.internal.client.ActivityClientHelper; +import io.temporal.internal.client.ActivityHeartbeatResponse; import io.temporal.internal.common.OptionsUtils; import io.temporal.internal.retryer.GrpcRetryer; import io.temporal.payload.context.ActivitySerializationContext; @@ -93,7 +94,7 @@ public void complete(@Nullable Object result) { .respondActivityTaskCompleted(request.build()), replyGrpcRetryerOptions); } catch (Exception e) { - processException(e); + throw wrapException(e); } } else { if (activityId == null) { @@ -115,7 +116,7 @@ public void complete(@Nullable Object result) { .respondActivityTaskCompletedById(request.build()), replyGrpcRetryerOptions); } catch (Exception e) { - processException(e); + throw wrapException(e); } } } @@ -168,16 +169,16 @@ public void fail(@Nonnull Throwable exception) { .respondActivityTaskFailedById(request), replyGrpcRetryerOptions); } catch (Exception e) { - processException(e); + throw wrapException(e); } } } @Override public void recordHeartbeat(@Nullable Object details) throws ActivityCompletionException { - if (taskToken != null) { - RecordActivityTaskHeartbeatResponse status; - try { + ActivityHeartbeatResponse status; + try { + if (taskToken != null) { status = grpcRetryer.retryWithResult( () -> @@ -189,23 +190,7 @@ public void recordHeartbeat(@Nullable Object details) throws ActivityCompletionE dataConverterWithActivityExecutionContext.toPayloads(details), metricsScope), replyGrpcRetryerOptions); - } catch (Exception e) { - processException(e); - return; - } - if (status.getCancelRequested()) { - throw new ActivityCanceledException(); - } else if (status.getActivityReset()) { - throw new ActivityResetException(); - } else if (status.getActivityPaused()) { - throw new ActivityPausedException(); - } - } else { - if (activityId == null) { - throw new IllegalArgumentException("Either activity id or task token are required"); - } - RecordActivityTaskHeartbeatByIdResponse status; - try { + } else { status = grpcRetryer.retryWithResult( () -> @@ -218,17 +203,16 @@ public void recordHeartbeat(@Nullable Object details) throws ActivityCompletionE dataConverterWithActivityExecutionContext.toPayloads(details), metricsScope), replyGrpcRetryerOptions); - } catch (Exception e) { - processException(e); - return; - } - if (status.getCancelRequested()) { - throw new ActivityCanceledException(); - } else if (status.getActivityReset()) { - throw new ActivityResetException(); - } else if (status.getActivityPaused()) { - throw new ActivityPausedException(); } + } catch (Exception e) { + throw wrapException(e); + } + if (status.getCancelRequested()) { + throw new ActivityCanceledException(); + } else if (status.getActivityReset()) { + throw new ActivityResetException(); + } else if (status.getActivityPaused()) { + throw new ActivityPausedException(); } } @@ -282,13 +266,13 @@ public void reportCancellation(@Nullable Object details) { } } - private void processException(Exception e) { + private ActivityCompletionException wrapException(Exception e) { if (e instanceof StatusRuntimeException) { StatusRuntimeException sre = (StatusRuntimeException) e; if (sre.getStatus().getCode() == Status.Code.NOT_FOUND) { - throw new ActivityNotExistsException(activityId, sre); + return new ActivityNotExistsException(activityId, sre); } } - throw new ActivityCompletionFailureException(activityId, e); + return new ActivityCompletionFailureException(activityId, e); } } From cb0373a8b7d6ba5cc92e70e0e55ab14c73006824 Mon Sep 17 00:00:00 2001 From: Dan Plyukhin Date: Mon, 24 Aug 2026 17:32:58 -0500 Subject: [PATCH 3/4] Remove tricky retry logic, which deserves its own PR --- .../ManualActivityCompletionClient.java | 4 +- .../client/ActivityHeartbeatResponse.java | 5 +- .../ManualActivityCompletionClientImpl.java | 36 ++++++------- ...anualActivityCompletionClientImplTest.java | 52 ++++++++++++++----- 4 files changed, 59 insertions(+), 38 deletions(-) diff --git a/temporal-sdk/src/main/java/io/temporal/activity/ManualActivityCompletionClient.java b/temporal-sdk/src/main/java/io/temporal/activity/ManualActivityCompletionClient.java index 34f30745ff..14013381c1 100644 --- a/temporal-sdk/src/main/java/io/temporal/activity/ManualActivityCompletionClient.java +++ b/temporal-sdk/src/main/java/io/temporal/activity/ManualActivityCompletionClient.java @@ -33,8 +33,8 @@ public interface ManualActivityCompletionClient { * @throws ActivityCompletionException if the server reports the activity was cancelled, reset, or * paused ({@link io.temporal.client.ActivityCanceledException}, {@link * io.temporal.client.ActivityResetException}, {@link - * io.temporal.client.ActivityPausedException}), or if the heartbeat RPC itself fails after - * retries ({@link io.temporal.client.ActivityCompletionFailureException}, {@link + * io.temporal.client.ActivityPausedException}), or if the heartbeat RPC fails ({@link + * io.temporal.client.ActivityCompletionFailureException}, {@link * io.temporal.client.ActivityNotExistsException}). */ void recordHeartbeat(@Nullable Object details) throws ActivityCompletionException; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHeartbeatResponse.java b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHeartbeatResponse.java index c17c1f77ee..a8547a629a 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHeartbeatResponse.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/ActivityHeartbeatResponse.java @@ -3,8 +3,9 @@ import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatByIdResponse; import io.temporal.api.workflowservice.v1.RecordActivityTaskHeartbeatResponse; -/** Container class to deduplicate {@link RecordActivityTaskHeartbeatByIdResponse} and - * {@link RecordActivityTaskHeartbeatResponse}. +/** + * Container class to deduplicate {@link RecordActivityTaskHeartbeatByIdResponse} and {@link + * RecordActivityTaskHeartbeatResponse}. */ public final class ActivityHeartbeatResponse { private final boolean cancelRequested; diff --git a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java index 30738a1467..22c5fea6bf 100644 --- a/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java +++ b/temporal-sdk/src/main/java/io/temporal/internal/client/external/ManualActivityCompletionClientImpl.java @@ -180,29 +180,23 @@ public void recordHeartbeat(@Nullable Object details) throws ActivityCompletionE try { if (taskToken != null) { status = - grpcRetryer.retryWithResult( - () -> - ActivityClientHelper.sendHeartbeatRequest( - service, - namespace, - identity, - taskToken, - dataConverterWithActivityExecutionContext.toPayloads(details), - metricsScope), - replyGrpcRetryerOptions); + ActivityClientHelper.sendHeartbeatRequest( + service, + namespace, + identity, + taskToken, + dataConverterWithActivityExecutionContext.toPayloads(details), + metricsScope); } else { status = - grpcRetryer.retryWithResult( - () -> - ActivityClientHelper.recordActivityTaskHeartbeatById( - service, - namespace, - identity, - execution, - activityId, - dataConverterWithActivityExecutionContext.toPayloads(details), - metricsScope), - replyGrpcRetryerOptions); + ActivityClientHelper.recordActivityTaskHeartbeatById( + service, + namespace, + identity, + execution, + activityId, + dataConverterWithActivityExecutionContext.toPayloads(details), + metricsScope); } } catch (Exception e) { throw wrapException(e); diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java index 73eac7545a..6e880b66e0 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java @@ -1,10 +1,10 @@ package io.temporal.internal.client.external; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.times; -import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import com.uber.m3.tally.NoopScope; @@ -16,6 +16,7 @@ import io.temporal.api.workflowservice.v1.WorkflowServiceGrpc; import io.temporal.client.ActivityCanceledException; import io.temporal.client.ActivityCompletionFailureException; +import io.temporal.client.ActivityNotExistsException; import io.temporal.client.ActivityPausedException; import io.temporal.client.ActivityResetException; import io.temporal.common.converter.GlobalDataConverter; @@ -70,6 +71,9 @@ private ManualActivityCompletionClientImpl clientWithActivityId() { null); } + // The tests below verify that exceptions from the heartbeat RPC are reported + // according to the documentation from {@link ManualActivityCompletionClient#recordHeartbeat(Object)}. + @Test public void cancelRequestedThrowsActivityCanceledExceptionNotSwallowed() { when(blockingStub.recordActivityTaskHeartbeat(any())) @@ -111,24 +115,46 @@ public void byIdCancelRequestedThrowsActivityCanceledExceptionNotSwallowed() { } @Test - public void transientRpcErrorIsRetriedThenSucceeds() { - when(blockingStub.recordActivityTaskHeartbeat(any())) - .thenThrow(new StatusRuntimeException(Status.RESOURCE_EXHAUSTED)) - .thenReturn(RecordActivityTaskHeartbeatResponse.getDefaultInstance()); + public void byIdActivityResetThrowsActivityResetExceptionNotSwallowed() { + when(blockingStub.recordActivityTaskHeartbeatById(any())) + .thenReturn( + RecordActivityTaskHeartbeatByIdResponse.newBuilder().setActivityReset(true).build()); + + assertThrows( + ActivityResetException.class, () -> clientWithActivityId().recordHeartbeat("details")); + } - // Should not throw: the transient error is retried and the second attempt succeeds. - clientWithTaskToken().recordHeartbeat("details"); + @Test + public void byIdActivityPausedThrowsActivityPausedExceptionNotSwallowed() { + when(blockingStub.recordActivityTaskHeartbeatById(any())) + .thenReturn( + RecordActivityTaskHeartbeatByIdResponse.newBuilder().setActivityPaused(true).build()); - verify(blockingStub, times(2)).recordActivityTaskHeartbeat(any()); + assertThrows( + ActivityPausedException.class, () -> clientWithActivityId().recordHeartbeat("details")); } @Test - public void nonTransientRpcErrorIsReportedAsActivityCompletionFailureException() { + public void notFoundIsReportedAsActivityNotExistsException() { when(blockingStub.recordActivityTaskHeartbeat(any())) - .thenThrow(new StatusRuntimeException(Status.INTERNAL)); + .thenThrow(new StatusRuntimeException(Status.NOT_FOUND)); assertThrows( - ActivityCompletionFailureException.class, - () -> clientWithTaskToken().recordHeartbeat("details")); + ActivityNotExistsException.class, () -> clientWithTaskToken().recordHeartbeat("details")); + } + + @Test + public void rpcErrorIsReportedAsActivityCompletionFailureException() { + when(blockingStub.recordActivityTaskHeartbeat(any())) + .thenThrow(new StatusRuntimeException(Status.INTERNAL)); + + ActivityCompletionFailureException failure = + assertThrows( + ActivityCompletionFailureException.class, + () -> clientWithTaskToken().recordHeartbeat("details")); + + assertTrue(failure.getCause() instanceof StatusRuntimeException); + assertEquals( + Status.Code.INTERNAL, ((StatusRuntimeException) failure.getCause()).getStatus().getCode()); } } From 4881e2b3b10e1409eb2487f6a0ce9ba4c9ae31c6 Mon Sep 17 00:00:00 2001 From: Dan Plyukhin Date: Mon, 24 Aug 2026 17:59:41 -0500 Subject: [PATCH 4/4] Remove comment that spotless didn't like --- .../external/ManualActivityCompletionClientImplTest.java | 3 --- 1 file changed, 3 deletions(-) diff --git a/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java index 6e880b66e0..3c253ccbbd 100644 --- a/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java +++ b/temporal-sdk/src/test/java/io/temporal/internal/client/external/ManualActivityCompletionClientImplTest.java @@ -71,9 +71,6 @@ private ManualActivityCompletionClientImpl clientWithActivityId() { null); } - // The tests below verify that exceptions from the heartbeat RPC are reported - // according to the documentation from {@link ManualActivityCompletionClient#recordHeartbeat(Object)}. - @Test public void cancelRequestedThrowsActivityCanceledExceptionNotSwallowed() { when(blockingStub.recordActivityTaskHeartbeat(any()))