#414

Committed 31 Oct 2026 05:46AM UTC coverage: 70.518% (+0.3%) from 70.246%

Build # #414

Build Type

push

github

Committed by

web-flow

Commit Message

Test sending of paginated workflow task completions (#3178)

Cover the paginated send path in WorkflowWorker, which previously had
no unit tests: intermediate pages are sent in order before the final
page, a completion is sent whole when the namespace does not support
pagination, buffer loss on any page resends every page from the
first, other errors and worker shutdown end the resend loop, and the
gRPC retryer never retries a lost buffer on its own.

Coverage Stats

8243 of 13748 branches covered (59.96%)

Branch coverage included in aggregate %.

32939 of 44651 relevant lines covered (73.77%)

0.74 hits per line

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

70.49

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

package io.temporal.internal.client;

import com.google.protobuf.ByteString;
import io.grpc.Deadline;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.temporal.api.common.v1.WorkflowExecution;
import io.temporal.api.enums.v1.EventType;
import io.temporal.api.history.v1.History;
import io.temporal.api.history.v1.HistoryEvent;
import io.temporal.api.nexus.v1.PropagatedSerializationContext;
import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest;
import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse;
import io.temporal.internal.client.external.GenericWorkflowClient;
import io.temporal.internal.common.WorkflowExecutionUtils;
import java.util.concurrent.*;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;

/** This class encapsulates async long poll logic of {@link RootWorkflowClientInvoker} */
final class WorkflowClientLongPollAsyncHelper {

  /**
   * Returns the history response whose single event closed the workflow, potentially waiting for
   * workflow or continue-as-new executions to complete. It carries the Nexus serialization context
   * of the closed run, if the server returned one on any page of that run. Completes exceptionally
   * with a {@link TimeoutException} if the workflow didn't complete within the specified timeout.
   */
  static CompletableFuture<GetWorkflowExecutionHistoryResponse> getInstanceCloseEventAsync(
      GenericWorkflowClient genericClient,
      WorkflowClientRequestFactory workflowClientHelper,
      @Nonnull WorkflowExecution workflowExecution,
      long timeout,
      TimeUnit unit) {
    Deadline longPollTimeoutDeadline = Deadline.after(timeout, unit);
    return getInstanceCloseEventAsync(
            genericClient,
            workflowClientHelper,
            workflowExecution,
            ByteString.EMPTY,
            null,
            longPollTimeoutDeadline)
        .handle(
            (closed, e) -> {
              if (e == null) {
                return closed;
              } else {
                throw handleException(e, longPollTimeoutDeadline, workflowExecution, timeout, unit);
              }
            });
  }

  private static CompletionException handleException(
      Throwable e,
      Deadline longPollTimeoutDeadline,
      @Nonnull WorkflowExecution workflowExecution,
      long timeout,
      TimeUnit unit) {
    if (e instanceof CompletionException) {
      Throwable cause = e.getCause();
      if (longPollTimeoutDeadline.isExpired()
          && cause instanceof StatusRuntimeException
          && Status.Code.DEADLINE_EXCEEDED.equals(
              ((StatusRuntimeException) cause).getStatus().getCode())) {
        // we want to form timeout exception only if the original deadline is indeed expired.
        // Otherwise, we should rethrow a raw DEADLINE_EXCEEDED. throwing TimeoutException
        // in this case will be highly misleading.
        return new CompletionException(
            WorkflowClientLongPollHelper.newTimeoutException(workflowExecution, timeout, unit));
      } else {
        return (CompletionException) e;
      }
    } else {
      return new CompletionException(e);
    }
  }

  /**
   * Returns an instance closing event, potentially waiting for workflow to complete.
   *
   * @param propagated the Nexus serialization context returned so far for the current run, carried
   *     across empty long polls of the same run.
   */
  private static CompletableFuture<GetWorkflowExecutionHistoryResponse> getInstanceCloseEventAsync(
      GenericWorkflowClient genericClient,
      WorkflowClientRequestFactory workflowClientHelper,
      final WorkflowExecution workflowExecution,
      ByteString pageToken,
      @Nullable PropagatedSerializationContext propagated,
      Deadline longPollTimeoutDeadline) {
    GetWorkflowExecutionHistoryRequest request =
        workflowClientHelper.newHistoryLongPollRequest(workflowExecution, pageToken);
    CompletableFuture<GetWorkflowExecutionHistoryResponse> response =
        genericClient.longPollHistoryAsync(request, longPollTimeoutDeadline);
    return response.thenComposeAsync(
        (r) -> {
          PropagatedSerializationContext runPropagated =
              r.hasPropagatedNexusSerializationContext()
                  ? r.getPropagatedNexusSerializationContext()
                  : propagated;
          History history = r.getHistory();
          if (history.getEventsCount() == 0) {
            // Empty poll returned
            ByteString nextPageToken =
                r.getNextPageToken().isEmpty() ? pageToken : r.getNextPageToken();
            return getInstanceCloseEventAsync(
                genericClient,
                workflowClientHelper,
                workflowExecution,
                nextPageToken,
                runPropagated,
                longPollTimeoutDeadline);
          }
          HistoryEvent event = history.getEvents(0); // should be only one event
          if (!WorkflowExecutionUtils.isWorkflowExecutionClosedEvent(event)) {
            throw new RuntimeException("Unexpected workflow execution closing event: " + event);
          }
          // Workflow called continueAsNew. Start polling the new generation with new runId.
          if (event.getEventType() == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW) {
            WorkflowExecution nextWorkflowExecution =
                WorkflowExecution.newBuilder()
                    .setWorkflowId(workflowExecution.getWorkflowId())
                    .setRunId(
                        event
                            .getWorkflowExecutionContinuedAsNewEventAttributes()
                            .getNewExecutionRunId())
                    .build();
            return getInstanceCloseEventAsync(
                genericClient,
                workflowClientHelper,
                nextWorkflowExecution,
                ByteString.EMPTY,
                null,
                longPollTimeoutDeadline);
          }
          return CompletableFuture.completedFuture(
              WorkflowClientLongPollHelper.withPropagatedNexusSerializationContext(
                  r, runPropagated));
        });
  }
}

1	package io.temporal.internal.client;
2
3	import com.google.protobuf.ByteString;
4	import io.grpc.Deadline;
5	import io.grpc.Status;
6	import io.grpc.StatusRuntimeException;
7	import io.temporal.api.common.v1.WorkflowExecution;
8	import io.temporal.api.enums.v1.EventType;
9	import io.temporal.api.history.v1.History;
10	import io.temporal.api.history.v1.HistoryEvent;
11	import io.temporal.api.nexus.v1.PropagatedSerializationContext;
12	import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryRequest;
13	import io.temporal.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse;
14	import io.temporal.internal.client.external.GenericWorkflowClient;
15	import io.temporal.internal.common.WorkflowExecutionUtils;
16	import java.util.concurrent.*;
17	import javax.annotation.Nonnull;
18	import javax.annotation.Nullable;
19
20	/** This class encapsulates async long poll logic of {@link RootWorkflowClientInvoker} */
21	final class WorkflowClientLongPollAsyncHelper {	×
22
23	/**
24	* Returns the history response whose single event closed the workflow, potentially waiting for
25	* workflow or continue-as-new executions to complete. It carries the Nexus serialization context
26	* of the closed run, if the server returned one on any page of that run. Completes exceptionally
27	* with a {@link TimeoutException} if the workflow didn't complete within the specified timeout.
28	*/
29	static CompletableFuture<GetWorkflowExecutionHistoryResponse> getInstanceCloseEventAsync(
30	GenericWorkflowClient genericClient,
31	WorkflowClientRequestFactory workflowClientHelper,
32	@Nonnull WorkflowExecution workflowExecution,
33	long timeout,
34	TimeUnit unit) {
35	Deadline longPollTimeoutDeadline = Deadline.after(timeout, unit);	1 ✔
36	return getInstanceCloseEventAsync(	1 ✔
37	genericClient,
38	workflowClientHelper,
39	workflowExecution,
40	ByteString.EMPTY,
41	null,
42	longPollTimeoutDeadline)
43	.handle(	1 ✔
44	(closed, e) -> {
45	if (e == null) {	1 ✔
46	return closed;	1 ✔
47	} else {
48	throw handleException(e, longPollTimeoutDeadline, workflowExecution, timeout, unit);	1 ✔
49	}
50	});
51	}
52
53	private static CompletionException handleException(
54	Throwable e,
55	Deadline longPollTimeoutDeadline,
56	@Nonnull WorkflowExecution workflowExecution,
57	long timeout,
58	TimeUnit unit) {
59	if (e instanceof CompletionException) {	1 !
60	Throwable cause = e.getCause();	1 ✔
61	if (longPollTimeoutDeadline.isExpired()	1 !
62	&& cause instanceof StatusRuntimeException
63	&& Status.Code.DEADLINE_EXCEEDED.equals(	1 !
64	((StatusRuntimeException) cause).getStatus().getCode())) {	1 ✔
65	// we want to form timeout exception only if the original deadline is indeed expired.
66	// Otherwise, we should rethrow a raw DEADLINE_EXCEEDED. throwing TimeoutException
67	// in this case will be highly misleading.
68	return new CompletionException(	1 ✔
69	WorkflowClientLongPollHelper.newTimeoutException(workflowExecution, timeout, unit));	1 ✔
70	} else {
71	return (CompletionException) e;	×
72	}
73	} else {
74	return new CompletionException(e);	×
75	}
76	}
77
78	/**
79	* Returns an instance closing event, potentially waiting for workflow to complete.
80	*
81	* @param propagated the Nexus serialization context returned so far for the current run, carried
82	* across empty long polls of the same run.
83	*/
84	private static CompletableFuture<GetWorkflowExecutionHistoryResponse> getInstanceCloseEventAsync(
85	GenericWorkflowClient genericClient,
86	WorkflowClientRequestFactory workflowClientHelper,
87	final WorkflowExecution workflowExecution,
88	ByteString pageToken,
89	@Nullable PropagatedSerializationContext propagated,
90	Deadline longPollTimeoutDeadline) {
91	GetWorkflowExecutionHistoryRequest request =	1 ✔
92	workflowClientHelper.newHistoryLongPollRequest(workflowExecution, pageToken);	1 ✔
93	CompletableFuture<GetWorkflowExecutionHistoryResponse> response =	1 ✔
94	genericClient.longPollHistoryAsync(request, longPollTimeoutDeadline);	1 ✔
95	return response.thenComposeAsync(	1 ✔
96	(r) -> {
97	PropagatedSerializationContext runPropagated =
98	r.hasPropagatedNexusSerializationContext()	1 ✔
99	? r.getPropagatedNexusSerializationContext()	1 ✔
100	: propagated;	1 ✔
101	History history = r.getHistory();	1 ✔
102	if (history.getEventsCount() == 0) {	1 ✔
103	// Empty poll returned
104	ByteString nextPageToken =
105	r.getNextPageToken().isEmpty() ? pageToken : r.getNextPageToken();	1 !
106	return getInstanceCloseEventAsync(	1 ✔
107	genericClient,
108	workflowClientHelper,
109	workflowExecution,
110	nextPageToken,
111	runPropagated,
112	longPollTimeoutDeadline);
113	}
114	HistoryEvent event = history.getEvents(0); // should be only one event	1 ✔
115	if (!WorkflowExecutionUtils.isWorkflowExecutionClosedEvent(event)) {	1 !
116	throw new RuntimeException("Unexpected workflow execution closing event: " + event);	×
117	}
118	// Workflow called continueAsNew. Start polling the new generation with new runId.
119	if (event.getEventType() == EventType.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW) {	1 !
120	WorkflowExecution nextWorkflowExecution =
121	WorkflowExecution.newBuilder()	×
122	.setWorkflowId(workflowExecution.getWorkflowId())	×
123	.setRunId(	×
124	event
125	.getWorkflowExecutionContinuedAsNewEventAttributes()	×
126	.getNewExecutionRunId())	×
127	.build();	×
128	return getInstanceCloseEventAsync(	×
129	genericClient,
130	workflowClientHelper,
131	nextWorkflowExecution,
132	ByteString.EMPTY,
133	null,
134	longPollTimeoutDeadline);
135	}
136	return CompletableFuture.completedFuture(	1 ✔
137	WorkflowClientLongPollHelper.withPropagatedNexusSerializationContext(	1 ✔
138	r, runPropagated));
139	});
140	}
141	}

temporalio / sdk-java / #414

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