#359

Committed 17 Aug 2026 10:04PM UTC coverage: 68.024% (-0.02%) from 68.042%

Build # #359

Build Type

push

github

Committed by

web-flow

Commit Message

Report non-retryable PayloadValidationError as BAD_REQUEST (#3009)

## What changed

A data converter can now signal that a Nexus operation's input is invalid by throwing a non-retryable `ApplicationFailure` of type `PayloadValidationError` while deserializing the input. `PayloadSerializer` translates such a failure into a `BAD_REQUEST` `HandlerException` with the message `invalid operation input`, retaining the original failure as its cause.

Previously any `ApplicationFailure` from the data converter propagated and was turned into an `INTERNAL` handler error by `NexusTaskHandlerImpl`, which callers retry — so a caller sending invalid input was retried until timeout instead of failing fast.

## Unchanged

- `ApplicationFailure` of any other type → propagated as before (`INTERNAL`)
- A **retryable** `PayloadValidationError` → propagated as before (`INTERNAL`); non-retryable is required
- `HandlerException` from the converter → passed through untouched
- `InvalidProtocolBufferException` / `DataConverterException` → `BAD_REQUEST` as before
- `serialize()` is untouched — `BAD_REQUEST` would be wrong for a result-encoding failure

## Tests

- `PayloadSerializerTest`: positive case asserting `BAD_REQUEST`, non-retryable, the wrapper message and the preserved cause; negative cases for a different error type and for a retryable `PayloadValidationError`.
- `OperationInputDeserializationErrorPropagationTest`: end-to-end through the test server, including a retryable case that proves the non-retryable guard by being retried.

## Cross-SDK

Part of a coordinated change; equivalent PRs exist for Go, TypeScript, Python and .NET. The wrapper message wording is aligned across SDKs, adapted to each SDK's message style.

Coverage Stats

7405 of 12974 branches covered (57.08%)

Branch coverage included in aggregate %.

5 of 5 new or added lines in 1 file covered. (100.0%)

17 existing lines in 6 files now uncovered.

30510 of 42764 relevant lines covered (71.35%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

55.88

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

package io.temporal.client;

import com.uber.m3.tally.Scope;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
import io.temporal.payload.context.ActivitySerializationContext;
import io.temporal.workflow.Functions;
import java.util.Optional;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

class ActivityCompletionClientImpl implements ActivityCompletionClient {

  private final ManualActivityCompletionClientFactory factory;
  private final Functions.Proc completionHandle;
  private final Scope metricsScope;
  private final @Nullable ActivitySerializationContext serializationContext;

  ActivityCompletionClientImpl(
      ManualActivityCompletionClientFactory manualActivityCompletionClientFactory,
      Functions.Proc completionHandle,
      Scope metricsScope,
      @Nullable ActivitySerializationContext serializationContext) {
    this.factory = manualActivityCompletionClientFactory;
    this.completionHandle = completionHandle;
    this.metricsScope = metricsScope;
    this.serializationContext = serializationContext;
  }

  @Override
  public <R> void complete(byte[] taskToken, R result) {
    try {
      factory.getClient(taskToken, metricsScope, serializationContext).complete(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <R> void complete(String workflowId, Optional<String> runId, String activityId, R result) {
    try {
      factory
          .getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)
          .complete(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <R> void completeStandalone(String activityId, Optional<String> activityRunId, R result) {
    try {
      factory
          .getClient(
              toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)
          .complete(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public void completeExceptionally(byte[] taskToken, Exception result) {
    try {
      factory.getClient(taskToken, metricsScope, serializationContext).fail(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public void completeExceptionally(
      String workflowId, Optional<String> runId, String activityId, Exception result) {
    try {
      factory
          .getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)
          .fail(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public void completeExceptionallyStandalone(
      String activityId, Optional<String> activityRunId, Exception result) {
    try {
      factory
          .getClient(
              toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)
          .fail(result);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <V> void reportCancellation(byte[] taskToken, V details) {
    try {
      factory.getClient(taskToken, metricsScope, serializationContext).reportCancellation(details);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <V> void reportCancellation(
      String workflowId, Optional<String> runId, String activityId, V details) {
    try {
      factory
          .getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)
          .reportCancellation(details);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <V> void reportCancellationStandalone(
      String activityId, Optional<String> activityRunId, V details) {
    try {
      factory
          .getClient(
              toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)
          .reportCancellation(details);
    } finally {
      completionHandle.apply();
    }
  }

  @Override
  public <V> void heartbeat(byte[] taskToken, V details) throws ActivityCompletionException {
    factory.getClient(taskToken, metricsScope).recordHeartbeat(details);
  }

  @Override
  public <V> void heartbeat(String workflowId, Optional<String> runId, String activityId, V details)
      throws ActivityCompletionException {
    factory
        .getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)
        .recordHeartbeat(details);
  }

  @Nonnull
  @Override
  public ActivityCompletionClient withContext(@Nonnull ActivitySerializationContext context) {
    return new ActivityCompletionClientImpl(factory, completionHandle, metricsScope, context);
  }

  @Override
  public <V> void heartbeatStandalone(String activityId, Optional<String> activityRunId, V details)
      throws ActivityCompletionException {
    factory
        .getClient(
            toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)
        .recordHeartbeat(details);
  }

  private static WorkflowExecution toExecution(String workflowId, Optional<String> runId) {
    return WorkflowExecution.newBuilder()
        .setWorkflowId(workflowId)
        .setRunId(runId.orElse(""))
        .build();
  }

  private static WorkflowExecution toStandaloneExecution(Optional<String> activityRunId) {
    // TODO: consider not using a WorkflowExecution here, but instead passing
    // workflow/activity and run IDs to the factory separately.
    return WorkflowExecution.newBuilder()
        .setWorkflowId("") // empty workflowId = standalone activity
        .setRunId(activityRunId.orElse(""))
        .build();
  }
}

1	package io.temporal.client;
2
3	import com.uber.m3.tally.Scope;
4	import io.temporal.api.common.v1.WorkflowExecution;
5	import io.temporal.internal.client.external.ManualActivityCompletionClientFactory;
6	import io.temporal.payload.context.ActivitySerializationContext;
7	import io.temporal.workflow.Functions;
8	import java.util.Optional;
9	import javax.annotation.Nonnull;
10	import javax.annotation.Nullable;
11
12	class ActivityCompletionClientImpl implements ActivityCompletionClient {
13
14	private final ManualActivityCompletionClientFactory factory;
15	private final Functions.Proc completionHandle;
16	private final Scope metricsScope;
17	private final @Nullable ActivitySerializationContext serializationContext;
18
19	ActivityCompletionClientImpl(
20	ManualActivityCompletionClientFactory manualActivityCompletionClientFactory,
21	Functions.Proc completionHandle,
22	Scope metricsScope,
23	@Nullable ActivitySerializationContext serializationContext) {	1✔
24	this.factory = manualActivityCompletionClientFactory;	1✔
25	this.completionHandle = completionHandle;	1✔
26	this.metricsScope = metricsScope;	1✔
27	this.serializationContext = serializationContext;	1✔
28	}	1✔
29
30	@Override
31	public <R> void complete(byte[] taskToken, R result) {
32	try {
33	factory.getClient(taskToken, metricsScope, serializationContext).complete(result);	1✔
34	} finally {
35	completionHandle.apply();	1✔
36	}
37	}	1✔
38
39	@Override
40	public <R> void complete(String workflowId, Optional<String> runId, String activityId, R result) {
41	try {
42	factory	×
43	.getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)	×
44	.complete(result);	×
45	} finally {
46	completionHandle.apply();	×
47	}
48	}	×
49
50	@Override
51	public <R> void completeStandalone(String activityId, Optional<String> activityRunId, R result) {
52	try {
53	factory	1✔
54	.getClient(	1✔
55	toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)	1✔
56	.complete(result);	1✔
57	} finally {
58	completionHandle.apply();	1✔
59	}
60	}	1✔
61
62	@Override
63	public void completeExceptionally(byte[] taskToken, Exception result) {
64	try {
65	factory.getClient(taskToken, metricsScope, serializationContext).fail(result);	×
66	} finally {
67	completionHandle.apply();	×
68	}
69	}	×
70
71	@Override
72	public void completeExceptionally(
73	String workflowId, Optional<String> runId, String activityId, Exception result) {
74	try {
75	factory	×
76	.getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)	×
77	.fail(result);	×
78	} finally {
79	completionHandle.apply();	×
80	}
81	}	×
82
83	@Override
84	public void completeExceptionallyStandalone(
85	String activityId, Optional<String> activityRunId, Exception result) {
86	try {
87	factory	1✔
88	.getClient(	1✔
89	toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)	1✔
90	.fail(result);	1✔
91	} finally {
92	completionHandle.apply();	1✔
93	}
94	}	1✔
95
96	@Override
97	public <V> void reportCancellation(byte[] taskToken, V details) {
98	try {
UNCOV 99	factory.getClient(taskToken, metricsScope, serializationContext).reportCancellation(details);	×
100	} finally {
UNCOV 101	completionHandle.apply();	×
102	}
UNCOV 103	}	×
104
105	@Override
106	public <V> void reportCancellation(
107	String workflowId, Optional<String> runId, String activityId, V details) {
108	try {
109	factory	×
110	.getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)	×
111	.reportCancellation(details);	×
112	} finally {
113	completionHandle.apply();	×
114	}
115	}	×
116
117	@Override
118	public <V> void reportCancellationStandalone(
119	String activityId, Optional<String> activityRunId, V details) {
120	try {
121	factory	1✔
122	.getClient(	1✔
123	toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)	1✔
124	.reportCancellation(details);	1✔
125	} finally {
126	completionHandle.apply();	1✔
127	}
128	}	1✔
129
130	@Override
131	public <V> void heartbeat(byte[] taskToken, V details) throws ActivityCompletionException {
132	factory.getClient(taskToken, metricsScope).recordHeartbeat(details);	1✔
133	}	1✔
134
135	@Override
136	public <V> void heartbeat(String workflowId, Optional<String> runId, String activityId, V details)
137	throws ActivityCompletionException {
138	factory	×
139	.getClient(toExecution(workflowId, runId), activityId, metricsScope, serializationContext)	×
140	.recordHeartbeat(details);	×
141	}	×
142
143	@Nonnull
144	@Override
145	public ActivityCompletionClient withContext(@Nonnull ActivitySerializationContext context) {
146	return new ActivityCompletionClientImpl(factory, completionHandle, metricsScope, context);	×
147	}
148
149	@Override
150	public <V> void heartbeatStandalone(String activityId, Optional<String> activityRunId, V details)
151	throws ActivityCompletionException {
152	factory	1✔
153	.getClient(	1✔
154	toStandaloneExecution(activityRunId), activityId, metricsScope, serializationContext)	1✔
155	.recordHeartbeat(details);	1✔
156	}	1✔
157
158	private static WorkflowExecution toExecution(String workflowId, Optional<String> runId) {
159	return WorkflowExecution.newBuilder()	×
160	.setWorkflowId(workflowId)	×
161	.setRunId(runId.orElse(""))	×
162	.build();	×
163	}
164
165	private static WorkflowExecution toStandaloneExecution(Optional<String> activityRunId) {
166	// TODO: consider not using a WorkflowExecution here, but instead passing
167	// workflow/activity and run IDs to the factory separately.
168	return WorkflowExecution.newBuilder()	1✔
169	.setWorkflowId("") // empty workflowId = standalone activity	1✔
170	.setRunId(activityRunId.orElse(""))	1✔
171	.build();	1✔
172	}
173	}

temporalio / sdk-java / #359

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous