#175

pending completion

Build # #175

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Worker / Build Id versioning (#1786)

Implement new worker build id based versioning feature

Run Details

236 of 236 new or added lines in 24 files covered. (100.0%)

18343 of 23697 relevant lines covered (77.41%)

0.81 hits per line

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

93.04

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

/*
 * Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
 *
 * Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
 *
 * Modifications copyright (C) 2017 Uber Technologies, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this material except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package io.temporal.internal.client;

import static io.temporal.internal.common.HeaderUtils.intoPayloadMap;

import io.grpc.Deadline;
import io.grpc.Status;
import io.grpc.StatusRuntimeException;
import io.temporal.api.common.v1.*;
import io.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage;
import io.temporal.api.enums.v1.WorkflowExecutionStatus;
import io.temporal.api.query.v1.WorkflowQuery;
import io.temporal.api.update.v1.*;
import io.temporal.api.workflowservice.v1.*;
import io.temporal.client.WorkflowClientOptions;
import io.temporal.client.WorkflowUpdateException;
import io.temporal.common.converter.DataConverter;
import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
import io.temporal.internal.client.external.GenericWorkflowClient;
import io.temporal.payload.context.WorkflowSerializationContext;
import io.temporal.worker.WorkflowTaskDispatchHandle;
import java.lang.reflect.Type;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import javax.annotation.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RootWorkflowClientInvoker implements WorkflowClientCallsInterceptor {
  private static final Logger log = LoggerFactory.getLogger(RootWorkflowClientInvoker.class);
  private static final long POLL_UPDATE_TIMEOUT_S = 60L;

  private final GenericWorkflowClient genericClient;
  private final WorkflowClientOptions clientOptions;
  private final EagerWorkflowTaskDispatcher eagerWorkflowTaskDispatcher;
  private final WorkflowClientRequestFactory requestsHelper;

  public RootWorkflowClientInvoker(
      GenericWorkflowClient genericClient,
      WorkflowClientOptions clientOptions,
      WorkerFactoryRegistry workerFactoryRegistry) {
    this.genericClient = genericClient;
    this.clientOptions = clientOptions;
    this.eagerWorkflowTaskDispatcher = new EagerWorkflowTaskDispatcher(workerFactoryRegistry);
    this.requestsHelper = new WorkflowClientRequestFactory(clientOptions);
  }

  @Override
  public WorkflowStartOutput start(WorkflowStartInput input) {
    DataConverter dataConverterWithWorkflowContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), input.getWorkflowId()));
    Optional<Payloads> inputArgs =
        dataConverterWithWorkflowContext.toPayloads(input.getArguments());

    @Nullable
    Memo memo =
        (input.getOptions().getMemo() != null)
            ? Memo.newBuilder()
                .putAllFields(
                    intoPayloadMap(clientOptions.getDataConverter(), input.getOptions().getMemo()))
                .build()
            : null;

    StartWorkflowExecutionRequest.Builder request =
        requestsHelper.newStartWorkflowExecutionRequest(
            input.getWorkflowId(),
            input.getWorkflowType(),
            input.getHeader(),
            input.getOptions(),
            inputArgs.orElse(null),
            memo);
    try (@Nullable WorkflowTaskDispatchHandle eagerDispatchHandle = obtainDispatchHandle(input)) {
      boolean requestEagerExecution = eagerDispatchHandle != null;
      request.setRequestEagerExecution(requestEagerExecution);
      StartWorkflowExecutionResponse response = genericClient.start(request.build());
      WorkflowExecution execution =
          WorkflowExecution.newBuilder()
              .setRunId(response.getRunId())
              .setWorkflowId(request.getWorkflowId())
              .build();
      @Nullable
      PollWorkflowTaskQueueResponse eagerWorkflowTask =
          requestEagerExecution && response.hasEagerWorkflowTask()
              ? response.getEagerWorkflowTask()
              : null;
      if (eagerWorkflowTask != null) {
        try {
          eagerDispatchHandle.dispatch(eagerWorkflowTask);
        } catch (Exception e) {
          // Any exception here is not expected, and it's a bug.
          // But we don't allow any exception from the dispatching to disrupt the control flow here,
          // the Client needs to get the execution back to matter what.
          // Inability to dispatch a WFT creates a latency issue, but it's not a failure of the
          // start itself
          log.error(
              "[BUG] Eager Workflow Task was received from the Server, but failed to be dispatched on the local worker",
              e);
        }
      }
      return new WorkflowStartOutput(execution);
    }
  }

  @Override
  public WorkflowSignalOutput signal(WorkflowSignalInput input) {
    SignalWorkflowExecutionRequest.Builder request =
        SignalWorkflowExecutionRequest.newBuilder()
            .setSignalName(input.getSignalName())
            .setWorkflowExecution(input.getWorkflowExecution())
            .setIdentity(clientOptions.getIdentity())
            .setNamespace(clientOptions.getNamespace());

    DataConverter dataConverterWitSignalContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));

    Optional<Payloads> inputArgs = dataConverterWitSignalContext.toPayloads(input.getArguments());
    inputArgs.ifPresent(request::setInput);
    genericClient.signal(request.build());
    return new WorkflowSignalOutput();
  }

  @Override
  public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInput input) {
    WorkflowStartInput workflowStartInput = input.getWorkflowStartInput();

    DataConverter dataConverterWithWorkflowContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), workflowStartInput.getWorkflowId()));
    Optional<Payloads> workflowInput =
        dataConverterWithWorkflowContext.toPayloads(workflowStartInput.getArguments());

    @Nullable
    Memo memo =
        (workflowStartInput.getOptions().getMemo() != null)
            ? Memo.newBuilder()
                .putAllFields(
                    intoPayloadMap(
                        clientOptions.getDataConverter(),
                        workflowStartInput.getOptions().getMemo()))
                .build()
            : null;

    StartWorkflowExecutionRequestOrBuilder startRequest =
        requestsHelper.newStartWorkflowExecutionRequest(
            workflowStartInput.getWorkflowId(),
            workflowStartInput.getWorkflowType(),
            workflowStartInput.getHeader(),
            workflowStartInput.getOptions(),
            workflowInput.orElse(null),
            memo);

    Optional<Payloads> signalInput =
        dataConverterWithWorkflowContext.toPayloads(input.getSignalArguments());
    SignalWithStartWorkflowExecutionRequest request =
        requestsHelper
            .newSignalWithStartWorkflowExecutionRequest(
                startRequest, input.getSignalName(), signalInput.orElse(null))
            .build();
    SignalWithStartWorkflowExecutionResponse response = genericClient.signalWithStart(request);
    WorkflowExecution execution =
        WorkflowExecution.newBuilder()
            .setRunId(response.getRunId())
            .setWorkflowId(request.getWorkflowId())
            .build();
    // TODO currently SignalWithStartWorkflowExecutionResponse doesn't have eagerWorkflowTask.
    //  We should wire it when it's implemented server-side.
    return new WorkflowSignalWithStartOutput(new WorkflowStartOutput(execution));
  }

  @Override
  public <R> GetResultOutput<R> getResult(GetResultInput<R> input) throws TimeoutException {
    DataConverter dataConverterWithWorkflowContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));
    Optional<Payloads> resultValue =
        WorkflowClientLongPollHelper.getWorkflowExecutionResult(
            genericClient,
            requestsHelper,
            input.getWorkflowExecution(),
            input.getWorkflowType(),
            dataConverterWithWorkflowContext,
            input.getTimeout(),
            input.getTimeoutUnit());
    return new GetResultOutput<>(
        convertResultPayloads(
            resultValue,
            input.getResultClass(),
            input.getResultType(),
            dataConverterWithWorkflowContext));
  }

  @Override
  public <R> GetResultAsyncOutput<R> getResultAsync(GetResultInput<R> input) {
    DataConverter dataConverterWithWorkflowContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));
    CompletableFuture<Optional<Payloads>> resultValue =
        WorkflowClientLongPollAsyncHelper.getWorkflowExecutionResultAsync(
            genericClient,
            requestsHelper,
            input.getWorkflowExecution(),
            input.getWorkflowType(),
            input.getTimeout(),
            input.getTimeoutUnit(),
            dataConverterWithWorkflowContext);
    return new GetResultAsyncOutput<>(
        resultValue.thenApply(
            payloads ->
                convertResultPayloads(
                    payloads,
                    input.getResultClass(),
                    input.getResultType(),
                    dataConverterWithWorkflowContext)));
  }

  @Override
  public <R> QueryOutput<R> query(QueryInput<R> input) {
    WorkflowQuery.Builder query = WorkflowQuery.newBuilder().setQueryType(input.getQueryType());
    DataConverter dataConverterWithWorkflowContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));

    Optional<Payloads> inputArgs =
        dataConverterWithWorkflowContext.toPayloads(input.getArguments());
    inputArgs.ifPresent(query::setQueryArgs);
    QueryWorkflowRequest request =
        QueryWorkflowRequest.newBuilder()
            .setNamespace(clientOptions.getNamespace())
            .setExecution(
                WorkflowExecution.newBuilder()
                    .setWorkflowId(input.getWorkflowExecution().getWorkflowId())
                    .setRunId(input.getWorkflowExecution().getRunId()))
            .setQuery(query)
            .setQueryRejectCondition(clientOptions.getQueryRejectCondition())
            .build();

    QueryWorkflowResponse result;
    result = genericClient.query(request);

    boolean queryRejected = result.hasQueryRejected();
    WorkflowExecutionStatus rejectStatus =
        queryRejected ? result.getQueryRejected().getStatus() : null;
    Optional<Payloads> queryResult =
        result.hasQueryResult() ? Optional.of(result.getQueryResult()) : Optional.empty();
    R resultValue =
        convertResultPayloads(
            queryResult,
            input.getResultClass(),
            input.getResultType(),
            dataConverterWithWorkflowContext);
    return new QueryOutput<>(rejectStatus, resultValue);
  }

  @Override
  public <R> StartUpdateOutput<R> startUpdate(StartUpdateInput<R> input) {
    DataConverter dataConverterWithWorkflowContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));

    Optional<Payloads> inputArgs =
        dataConverterWithWorkflowContext.toPayloads(input.getArguments());
    Input.Builder updateInput = Input.newBuilder().setName(input.getUpdateName());
    inputArgs.ifPresent(updateInput::setArgs);

    Request request =
        Request.newBuilder()
            .setMeta(
                Meta.newBuilder()
                    .setUpdateId(input.getUpdateId())
                    .setIdentity(clientOptions.getIdentity()))
            .setInput(updateInput)
            .build();
    UpdateWorkflowExecutionRequest updateRequest =
        UpdateWorkflowExecutionRequest.newBuilder()
            .setNamespace(clientOptions.getNamespace())
            .setWaitPolicy(input.getWaitPolicy())
            .setWorkflowExecution(
                WorkflowExecution.newBuilder()
                    .setWorkflowId(input.getWorkflowExecution().getWorkflowId())
                    .setRunId(input.getWorkflowExecution().getRunId()))
            .setFirstExecutionRunId(input.getFirstExecutionRunId())
            .setRequest(request)
            .build();
    Deadline pollTimeoutDeadline = Deadline.after(POLL_UPDATE_TIMEOUT_S, TimeUnit.SECONDS);
    UpdateWorkflowExecutionResponse result =
        genericClient.update(updateRequest, pollTimeoutDeadline);

    if (result.hasOutcome()) {
      switch (result.getOutcome().getValueCase()) {
        case SUCCESS:
          Optional<Payloads> updateResult = Optional.of(result.getOutcome().getSuccess());
          R resultValue =
              convertResultPayloads(
                  updateResult,
                  input.getResultClass(),
                  input.getResultType(),
                  dataConverterWithWorkflowContext);
          return new StartUpdateOutput<R>(result.getUpdateRef(), true, resultValue);
        case FAILURE:
          throw new WorkflowUpdateException(
              result.getUpdateRef().getWorkflowExecution(),
              result.getUpdateRef().getUpdateId(),
              input.getUpdateName(),
              dataConverterWithWorkflowContext.failureToException(
                  result.getOutcome().getFailure()));
        default:
          throw new RuntimeException(
              "Received unexpected outcome from update request: "
                  + result.getOutcome().getValueCase());
      }
    } else {
      return new StartUpdateOutput<R>(result.getUpdateRef(), false, null);
    }
  }

  @Override
  public <R> PollWorkflowUpdateOutput<R> pollWorkflowUpdate(PollWorkflowUpdateInput<R> input) {
    DataConverter dataConverterWithWorkflowContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));

    UpdateRef update =
        UpdateRef.newBuilder()
            .setWorkflowExecution(input.getWorkflowExecution())
            .setUpdateId(input.getUpdateId())
            .build();

    WaitPolicy waitPolicy =
        WaitPolicy.newBuilder()
            .setLifecycleStage(
                UpdateWorkflowExecutionLifecycleStage
                    .UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED)
            .build();

    PollWorkflowExecutionUpdateRequest pollUpdateRequest =
        PollWorkflowExecutionUpdateRequest.newBuilder()
            .setNamespace(clientOptions.getNamespace())
            .setIdentity(clientOptions.getIdentity())
            .setUpdateRef(update)
            .setWaitPolicy(waitPolicy)
            .build();

    CompletableFuture<PollWorkflowExecutionUpdateResponse> future = new CompletableFuture<>();

    Deadline pollTimeoutDeadline = Deadline.after(input.getTimeout(), input.getTimeoutUnit());
    pollWorkflowUpdateHelper(future, pollUpdateRequest, pollTimeoutDeadline);
    return new PollWorkflowUpdateOutput(
        future.thenApply(
            (result) -> {
              if (result.hasOutcome()) {
                switch (result.getOutcome().getValueCase()) {
                  case SUCCESS:
                    Optional<Payloads> updateResult = Optional.of(result.getOutcome().getSuccess());
                    R resultValue =
                        convertResultPayloads(
                            updateResult,
                            input.getResultClass(),
                            input.getResultType(),
                            dataConverterWithWorkflowContext);
                    return resultValue;
                  case FAILURE:
                    throw new WorkflowUpdateException(
                        input.getWorkflowExecution(),
                        input.getUpdateId(),
                        input.getUpdateName(),
                        dataConverterWithWorkflowContext.failureToException(
                            result.getOutcome().getFailure()));
                  default:
                    throw new RuntimeException(
                        "Received unexpected outcome from poll update request: "
                            + result.getOutcome().getValueCase());
                }
              }
              throw new RuntimeException("Received no outcome from server");
            }));
  }

  private void pollWorkflowUpdateHelper(
      CompletableFuture<PollWorkflowExecutionUpdateResponse> resultCF,
      PollWorkflowExecutionUpdateRequest request,
      Deadline deadline) {

    Deadline pollTimeoutDeadline =
        Deadline.after(POLL_UPDATE_TIMEOUT_S, TimeUnit.SECONDS).minimum(deadline);
    genericClient
        .pollUpdateAsync(request, pollTimeoutDeadline)
        .whenComplete(
            (r, e) -> {
              if ((e instanceof StatusRuntimeException
                      && ((StatusRuntimeException) e).getStatus().getCode()
                          == Status.Code.DEADLINE_EXCEEDED)
                  || pollTimeoutDeadline.isExpired()
                  || (e == null && !r.hasOutcome())) {
                // if the request has timed out, stop retrying
                if (!deadline.isExpired()) {
                  pollWorkflowUpdateHelper(resultCF, request, deadline);
                } else {
                  resultCF.completeExceptionally(
                      new TimeoutException(
                          "WorkflowId="
                              + request.getUpdateRef().getWorkflowExecution().getWorkflowId()
                              + ", runId="
                              + request.getUpdateRef().getWorkflowExecution().getRunId()
                              + ", updateId="
                              + request.getUpdateRef().getUpdateId()));
                }
              } else if (e != null) {
                resultCF.completeExceptionally(e);
              } else {
                resultCF.complete(r);
              }
            });
  }

  @Override
  public CancelOutput cancel(CancelInput input) {
    RequestCancelWorkflowExecutionRequest.Builder request =
        RequestCancelWorkflowExecutionRequest.newBuilder()
            .setRequestId(UUID.randomUUID().toString())
            .setWorkflowExecution(input.getWorkflowExecution())
            .setNamespace(clientOptions.getNamespace())
            .setIdentity(clientOptions.getIdentity());
    genericClient.requestCancel(request.build());
    return new CancelOutput();
  }

  @Override
  public TerminateOutput terminate(TerminateInput input) {
    TerminateWorkflowExecutionRequest.Builder request =
        TerminateWorkflowExecutionRequest.newBuilder()
            .setNamespace(clientOptions.getNamespace())
            .setWorkflowExecution(input.getWorkflowExecution());
    if (input.getReason() != null) {
      request.setReason(input.getReason());
    }
    DataConverter dataConverterWithWorkflowContext =
        clientOptions
            .getDataConverter()
            .withContext(
                new WorkflowSerializationContext(
                    clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));
    Optional<Payloads> payloads = dataConverterWithWorkflowContext.toPayloads(input.getDetails());
    payloads.ifPresent(request::setDetails);
    genericClient.terminate(request.build());
    return new TerminateOutput();
  }

  private static <R> R convertResultPayloads(
      Optional<Payloads> resultValue,
      Class<R> resultClass,
      Type resultType,
      DataConverter dataConverter) {
    return dataConverter.fromPayloads(0, resultValue, resultClass, resultType);
  }

  /**
   * @return a handle to dispatch the eager workflow task. {@code null} if an eager execution is
   *     disabled through {@link io.temporal.client.WorkflowOptions} or the worker
   *     <ul>
   *       <li>is activity only worker
   *       <li>not started, shutdown or paused
   *       <li>doesn't have an executor slot available
   *     </ul>
   */
  @Nullable
  private WorkflowTaskDispatchHandle obtainDispatchHandle(WorkflowStartInput input) {
    if (input.getOptions().isDisableEagerExecution()) {
      return null;
    }
    return eagerWorkflowTaskDispatcher.tryGetLocalDispatchHandler(input);
  }
}

1	/*
2	* Copyright (C) 2022 Temporal Technologies, Inc. All Rights Reserved.
3	*
4	* Copyright (C) 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
5	*
6	* Modifications copyright (C) 2017 Uber Technologies, Inc.
7	*
8	* Licensed under the Apache License, Version 2.0 (the "License");
9	* you may not use this material except in compliance with the License.
10	* You may obtain a copy of the License at
11	*
12	* http://www.apache.org/licenses/LICENSE-2.0
13	*
14	* Unless required by applicable law or agreed to in writing, software
15	* distributed under the License is distributed on an "AS IS" BASIS,
16	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17	* See the License for the specific language governing permissions and
18	* limitations under the License.
19	*/
20
21	package io.temporal.internal.client;
22
23	import static io.temporal.internal.common.HeaderUtils.intoPayloadMap;
24
25	import io.grpc.Deadline;
26	import io.grpc.Status;
27	import io.grpc.StatusRuntimeException;
28	import io.temporal.api.common.v1.*;
29	import io.temporal.api.enums.v1.UpdateWorkflowExecutionLifecycleStage;
30	import io.temporal.api.enums.v1.WorkflowExecutionStatus;
31	import io.temporal.api.query.v1.WorkflowQuery;
32	import io.temporal.api.update.v1.*;
33	import io.temporal.api.workflowservice.v1.*;
34	import io.temporal.client.WorkflowClientOptions;
35	import io.temporal.client.WorkflowUpdateException;
36	import io.temporal.common.converter.DataConverter;
37	import io.temporal.common.interceptors.WorkflowClientCallsInterceptor;
38	import io.temporal.internal.client.external.GenericWorkflowClient;
39	import io.temporal.payload.context.WorkflowSerializationContext;
40	import io.temporal.worker.WorkflowTaskDispatchHandle;
41	import java.lang.reflect.Type;
42	import java.util.*;
43	import java.util.concurrent.CompletableFuture;
44	import java.util.concurrent.TimeUnit;
45	import java.util.concurrent.TimeoutException;
46	import javax.annotation.Nullable;
47	import org.slf4j.Logger;
48	import org.slf4j.LoggerFactory;
49
50	public class RootWorkflowClientInvoker implements WorkflowClientCallsInterceptor {
51	private static final Logger log = LoggerFactory.getLogger(RootWorkflowClientInvoker.class);	1✔
52	private static final long POLL_UPDATE_TIMEOUT_S = 60L;
53
54	private final GenericWorkflowClient genericClient;
55	private final WorkflowClientOptions clientOptions;
56	private final EagerWorkflowTaskDispatcher eagerWorkflowTaskDispatcher;
57	private final WorkflowClientRequestFactory requestsHelper;
58
59	public RootWorkflowClientInvoker(
60	GenericWorkflowClient genericClient,
61	WorkflowClientOptions clientOptions,
62	WorkerFactoryRegistry workerFactoryRegistry) {	1✔
63	this.genericClient = genericClient;	1✔
64	this.clientOptions = clientOptions;	1✔
65	this.eagerWorkflowTaskDispatcher = new EagerWorkflowTaskDispatcher(workerFactoryRegistry);	1✔
66	this.requestsHelper = new WorkflowClientRequestFactory(clientOptions);	1✔
67	}	1✔
68
69	@Override
70	public WorkflowStartOutput start(WorkflowStartInput input) {
71	DataConverter dataConverterWithWorkflowContext =	1✔
72	clientOptions
73	.getDataConverter()	1✔
74	.withContext(	1✔
75	new WorkflowSerializationContext(
76	clientOptions.getNamespace(), input.getWorkflowId()));	1✔
77	Optional<Payloads> inputArgs =	1✔
78	dataConverterWithWorkflowContext.toPayloads(input.getArguments());	1✔
79
80	@Nullable
81	Memo memo =
82	(input.getOptions().getMemo() != null)	1✔
83	? Memo.newBuilder()	1✔
84	.putAllFields(	1✔
85	intoPayloadMap(clientOptions.getDataConverter(), input.getOptions().getMemo()))	1✔
86	.build()	1✔
87	: null;	1✔
88
89	StartWorkflowExecutionRequest.Builder request =	1✔
90	requestsHelper.newStartWorkflowExecutionRequest(	1✔
91	input.getWorkflowId(),	1✔
92	input.getWorkflowType(),	1✔
93	input.getHeader(),	1✔
94	input.getOptions(),	1✔
95	inputArgs.orElse(null),	1✔
96	memo);
97	try (@Nullable WorkflowTaskDispatchHandle eagerDispatchHandle = obtainDispatchHandle(input)) {	1✔
98	boolean requestEagerExecution = eagerDispatchHandle != null;	1✔
99	request.setRequestEagerExecution(requestEagerExecution);	1✔
100	StartWorkflowExecutionResponse response = genericClient.start(request.build());	1✔
101	WorkflowExecution execution =
102	WorkflowExecution.newBuilder()	1✔
103	.setRunId(response.getRunId())	1✔
104	.setWorkflowId(request.getWorkflowId())	1✔
105	.build();	1✔
106	@Nullable
107	PollWorkflowTaskQueueResponse eagerWorkflowTask =
108	requestEagerExecution && response.hasEagerWorkflowTask()	1✔
109	? response.getEagerWorkflowTask()	1✔
110	: null;	1✔
111	if (eagerWorkflowTask != null) {	1✔
112	try {
113	eagerDispatchHandle.dispatch(eagerWorkflowTask);	1✔
114	} catch (Exception e) {	×
115	// Any exception here is not expected, and it's a bug.
116	// But we don't allow any exception from the dispatching to disrupt the control flow here,
117	// the Client needs to get the execution back to matter what.
118	// Inability to dispatch a WFT creates a latency issue, but it's not a failure of the
119	// start itself
120	log.error(	×
121	"[BUG] Eager Workflow Task was received from the Server, but failed to be dispatched on the local worker",
122	e);
123	}	1✔
124	}
125	return new WorkflowStartOutput(execution);	1✔
126	}
127	}
128
129	@Override
130	public WorkflowSignalOutput signal(WorkflowSignalInput input) {
131	SignalWorkflowExecutionRequest.Builder request =
132	SignalWorkflowExecutionRequest.newBuilder()	1✔
133	.setSignalName(input.getSignalName())	1✔
134	.setWorkflowExecution(input.getWorkflowExecution())	1✔
135	.setIdentity(clientOptions.getIdentity())	1✔
136	.setNamespace(clientOptions.getNamespace());	1✔
137
138	DataConverter dataConverterWitSignalContext =	1✔
139	clientOptions
140	.getDataConverter()	1✔
141	.withContext(	1✔
142	new WorkflowSerializationContext(
143	clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));	1✔
144
145	Optional<Payloads> inputArgs = dataConverterWitSignalContext.toPayloads(input.getArguments());	1✔
146	inputArgs.ifPresent(request::setInput);	1✔
147	genericClient.signal(request.build());	1✔
148	return new WorkflowSignalOutput();	1✔
149	}
150
151	@Override
152	public WorkflowSignalWithStartOutput signalWithStart(WorkflowSignalWithStartInput input) {
153	WorkflowStartInput workflowStartInput = input.getWorkflowStartInput();	1✔
154
155	DataConverter dataConverterWithWorkflowContext =	1✔
156	clientOptions
157	.getDataConverter()	1✔
158	.withContext(	1✔
159	new WorkflowSerializationContext(
160	clientOptions.getNamespace(), workflowStartInput.getWorkflowId()));	1✔
161	Optional<Payloads> workflowInput =	1✔
162	dataConverterWithWorkflowContext.toPayloads(workflowStartInput.getArguments());	1✔
163
164	@Nullable
165	Memo memo =
166	(workflowStartInput.getOptions().getMemo() != null)	1✔
167	? Memo.newBuilder()	×
168	.putAllFields(	×
169	intoPayloadMap(	×
170	clientOptions.getDataConverter(),	×
171	workflowStartInput.getOptions().getMemo()))	×
172	.build()	×
173	: null;	1✔
174
175	StartWorkflowExecutionRequestOrBuilder startRequest =	1✔
176	requestsHelper.newStartWorkflowExecutionRequest(	1✔
177	workflowStartInput.getWorkflowId(),	1✔
178	workflowStartInput.getWorkflowType(),	1✔
179	workflowStartInput.getHeader(),	1✔
180	workflowStartInput.getOptions(),	1✔
181	workflowInput.orElse(null),	1✔
182	memo);
183
184	Optional<Payloads> signalInput =	1✔
185	dataConverterWithWorkflowContext.toPayloads(input.getSignalArguments());	1✔
186	SignalWithStartWorkflowExecutionRequest request =	1✔
187	requestsHelper
188	.newSignalWithStartWorkflowExecutionRequest(	1✔
189	startRequest, input.getSignalName(), signalInput.orElse(null))	1✔
190	.build();	1✔
191	SignalWithStartWorkflowExecutionResponse response = genericClient.signalWithStart(request);	1✔
192	WorkflowExecution execution =
193	WorkflowExecution.newBuilder()	1✔
194	.setRunId(response.getRunId())	1✔
195	.setWorkflowId(request.getWorkflowId())	1✔
196	.build();	1✔
197	// TODO currently SignalWithStartWorkflowExecutionResponse doesn't have eagerWorkflowTask.
198	// We should wire it when it's implemented server-side.
199	return new WorkflowSignalWithStartOutput(new WorkflowStartOutput(execution));	1✔
200	}
201
202	@Override
203	public <R> GetResultOutput<R> getResult(GetResultInput<R> input) throws TimeoutException {
204	DataConverter dataConverterWithWorkflowContext =	1✔
205	clientOptions
206	.getDataConverter()	1✔
207	.withContext(	1✔
208	new WorkflowSerializationContext(
209	clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));	1✔
210	Optional<Payloads> resultValue =	1✔
211	WorkflowClientLongPollHelper.getWorkflowExecutionResult(	1✔
212	genericClient,
213	requestsHelper,
214	input.getWorkflowExecution(),	1✔
215	input.getWorkflowType(),	1✔
216	dataConverterWithWorkflowContext,
217	input.getTimeout(),	1✔
218	input.getTimeoutUnit());	1✔
219	return new GetResultOutput<>(	1✔
220	convertResultPayloads(	1✔
221	resultValue,
222	input.getResultClass(),	1✔
223	input.getResultType(),	1✔
224	dataConverterWithWorkflowContext));
225	}
226
227	@Override
228	public <R> GetResultAsyncOutput<R> getResultAsync(GetResultInput<R> input) {
229	DataConverter dataConverterWithWorkflowContext =	1✔
230	clientOptions
231	.getDataConverter()	1✔
232	.withContext(	1✔
233	new WorkflowSerializationContext(
234	clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));	1✔
235	CompletableFuture<Optional<Payloads>> resultValue =	1✔
236	WorkflowClientLongPollAsyncHelper.getWorkflowExecutionResultAsync(	1✔
237	genericClient,
238	requestsHelper,
239	input.getWorkflowExecution(),	1✔
240	input.getWorkflowType(),	1✔
241	input.getTimeout(),	1✔
242	input.getTimeoutUnit(),	1✔
243	dataConverterWithWorkflowContext);
244	return new GetResultAsyncOutput<>(	1✔
245	resultValue.thenApply(	1✔
246	payloads ->
247	convertResultPayloads(	1✔
248	payloads,
249	input.getResultClass(),	1✔
250	input.getResultType(),	1✔
251	dataConverterWithWorkflowContext)));
252	}
253
254	@Override
255	public <R> QueryOutput<R> query(QueryInput<R> input) {
256	WorkflowQuery.Builder query = WorkflowQuery.newBuilder().setQueryType(input.getQueryType());	1✔
257	DataConverter dataConverterWithWorkflowContext =	1✔
258	clientOptions
259	.getDataConverter()	1✔
260	.withContext(	1✔
261	new WorkflowSerializationContext(
262	clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));	1✔
263
264	Optional<Payloads> inputArgs =	1✔
265	dataConverterWithWorkflowContext.toPayloads(input.getArguments());	1✔
266	inputArgs.ifPresent(query::setQueryArgs);	1✔
267	QueryWorkflowRequest request =
268	QueryWorkflowRequest.newBuilder()	1✔
269	.setNamespace(clientOptions.getNamespace())	1✔
270	.setExecution(	1✔
271	WorkflowExecution.newBuilder()	1✔
272	.setWorkflowId(input.getWorkflowExecution().getWorkflowId())	1✔
273	.setRunId(input.getWorkflowExecution().getRunId()))	1✔
274	.setQuery(query)	1✔
275	.setQueryRejectCondition(clientOptions.getQueryRejectCondition())	1✔
276	.build();	1✔
277
278	QueryWorkflowResponse result;
279	result = genericClient.query(request);	1✔
280
281	boolean queryRejected = result.hasQueryRejected();	1✔
282	WorkflowExecutionStatus rejectStatus =
283	queryRejected ? result.getQueryRejected().getStatus() : null;	1✔
284	Optional<Payloads> queryResult =
285	result.hasQueryResult() ? Optional.of(result.getQueryResult()) : Optional.empty();	1✔
286	R resultValue =	1✔
287	convertResultPayloads(	1✔
288	queryResult,
289	input.getResultClass(),	1✔
290	input.getResultType(),	1✔
291	dataConverterWithWorkflowContext);
292	return new QueryOutput<>(rejectStatus, resultValue);	1✔
293	}
294
295	@Override
296	public <R> StartUpdateOutput<R> startUpdate(StartUpdateInput<R> input) {
297	DataConverter dataConverterWithWorkflowContext =	1✔
298	clientOptions
299	.getDataConverter()	1✔
300	.withContext(	1✔
301	new WorkflowSerializationContext(
302	clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));	1✔
303
304	Optional<Payloads> inputArgs =	1✔
305	dataConverterWithWorkflowContext.toPayloads(input.getArguments());	1✔
306	Input.Builder updateInput = Input.newBuilder().setName(input.getUpdateName());	1✔
307	inputArgs.ifPresent(updateInput::setArgs);	1✔
308
309	Request request =
310	Request.newBuilder()	1✔
311	.setMeta(	1✔
312	Meta.newBuilder()	1✔
313	.setUpdateId(input.getUpdateId())	1✔
314	.setIdentity(clientOptions.getIdentity()))	1✔
315	.setInput(updateInput)	1✔
316	.build();	1✔
317	UpdateWorkflowExecutionRequest updateRequest =
318	UpdateWorkflowExecutionRequest.newBuilder()	1✔
319	.setNamespace(clientOptions.getNamespace())	1✔
320	.setWaitPolicy(input.getWaitPolicy())	1✔
321	.setWorkflowExecution(	1✔
322	WorkflowExecution.newBuilder()	1✔
323	.setWorkflowId(input.getWorkflowExecution().getWorkflowId())	1✔
324	.setRunId(input.getWorkflowExecution().getRunId()))	1✔
325	.setFirstExecutionRunId(input.getFirstExecutionRunId())	1✔
326	.setRequest(request)	1✔
327	.build();	1✔
328	Deadline pollTimeoutDeadline = Deadline.after(POLL_UPDATE_TIMEOUT_S, TimeUnit.SECONDS);	1✔
329	UpdateWorkflowExecutionResponse result =	1✔
330	genericClient.update(updateRequest, pollTimeoutDeadline);	1✔
331
332	if (result.hasOutcome()) {	1✔
333	switch (result.getOutcome().getValueCase()) {	1✔
334	case SUCCESS:
335	Optional<Payloads> updateResult = Optional.of(result.getOutcome().getSuccess());	1✔
336	R resultValue =	1✔
337	convertResultPayloads(	1✔
338	updateResult,
339	input.getResultClass(),	1✔
340	input.getResultType(),	1✔
341	dataConverterWithWorkflowContext);
342	return new StartUpdateOutput<R>(result.getUpdateRef(), true, resultValue);	1✔
343	case FAILURE:
344	throw new WorkflowUpdateException(	1✔
345	result.getUpdateRef().getWorkflowExecution(),	1✔
346	result.getUpdateRef().getUpdateId(),	1✔
347	input.getUpdateName(),	1✔
348	dataConverterWithWorkflowContext.failureToException(	1✔
349	result.getOutcome().getFailure()));	1✔
350	default:
351	throw new RuntimeException(	×
352	"Received unexpected outcome from update request: "
353	+ result.getOutcome().getValueCase());	×
354	}
355	} else {
356	return new StartUpdateOutput<R>(result.getUpdateRef(), false, null);	1✔
357	}
358	}
359
360	@Override
361	public <R> PollWorkflowUpdateOutput<R> pollWorkflowUpdate(PollWorkflowUpdateInput<R> input) {
362	DataConverter dataConverterWithWorkflowContext =	1✔
363	clientOptions
364	.getDataConverter()	1✔
365	.withContext(	1✔
366	new WorkflowSerializationContext(
367	clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));	1✔
368
369	UpdateRef update =
370	UpdateRef.newBuilder()	1✔
371	.setWorkflowExecution(input.getWorkflowExecution())	1✔
372	.setUpdateId(input.getUpdateId())	1✔
373	.build();	1✔
374
375	WaitPolicy waitPolicy =
376	WaitPolicy.newBuilder()	1✔
377	.setLifecycleStage(	1✔
378	UpdateWorkflowExecutionLifecycleStage
379	.UPDATE_WORKFLOW_EXECUTION_LIFECYCLE_STAGE_COMPLETED)
380	.build();	1✔
381
382	PollWorkflowExecutionUpdateRequest pollUpdateRequest =
383	PollWorkflowExecutionUpdateRequest.newBuilder()	1✔
384	.setNamespace(clientOptions.getNamespace())	1✔
385	.setIdentity(clientOptions.getIdentity())	1✔
386	.setUpdateRef(update)	1✔
387	.setWaitPolicy(waitPolicy)	1✔
388	.build();	1✔
389
390	CompletableFuture<PollWorkflowExecutionUpdateResponse> future = new CompletableFuture<>();	1✔
391
392	Deadline pollTimeoutDeadline = Deadline.after(input.getTimeout(), input.getTimeoutUnit());	1✔
393	pollWorkflowUpdateHelper(future, pollUpdateRequest, pollTimeoutDeadline);	1✔
394	return new PollWorkflowUpdateOutput(	1✔
395	future.thenApply(	1✔
396	(result) -> {
397	if (result.hasOutcome()) {	1✔
398	switch (result.getOutcome().getValueCase()) {	1✔
399	case SUCCESS:
400	Optional<Payloads> updateResult = Optional.of(result.getOutcome().getSuccess());	1✔
401	R resultValue =	1✔
402	convertResultPayloads(	1✔
403	updateResult,
404	input.getResultClass(),	1✔
405	input.getResultType(),	1✔
406	dataConverterWithWorkflowContext);
407	return resultValue;	1✔
408	case FAILURE:
409	throw new WorkflowUpdateException(	×
410	input.getWorkflowExecution(),	×
411	input.getUpdateId(),	×
412	input.getUpdateName(),	×
413	dataConverterWithWorkflowContext.failureToException(	×
414	result.getOutcome().getFailure()));	×
415	default:
416	throw new RuntimeException(	×
417	"Received unexpected outcome from poll update request: "
418	+ result.getOutcome().getValueCase());	×
419	}
420	}
421	throw new RuntimeException("Received no outcome from server");	×
422	}));
423	}
424
425	private void pollWorkflowUpdateHelper(
426	CompletableFuture<PollWorkflowExecutionUpdateResponse> resultCF,
427	PollWorkflowExecutionUpdateRequest request,
428	Deadline deadline) {
429
430	Deadline pollTimeoutDeadline =	1✔
431	Deadline.after(POLL_UPDATE_TIMEOUT_S, TimeUnit.SECONDS).minimum(deadline);	1✔
432	genericClient	1✔
433	.pollUpdateAsync(request, pollTimeoutDeadline)	1✔
434	.whenComplete(	1✔
435	(r, e) -> {
436	if ((e instanceof StatusRuntimeException	1✔
437	&& ((StatusRuntimeException) e).getStatus().getCode()	1✔
438	== Status.Code.DEADLINE_EXCEEDED)
439	\|\| pollTimeoutDeadline.isExpired()	1✔
440	\|\| (e == null && !r.hasOutcome())) {	1✔
441	// if the request has timed out, stop retrying
442	if (!deadline.isExpired()) {	1✔
443	pollWorkflowUpdateHelper(resultCF, request, deadline);	1✔
444	} else {
445	resultCF.completeExceptionally(	1✔
446	new TimeoutException(
447	"WorkflowId="
448	+ request.getUpdateRef().getWorkflowExecution().getWorkflowId()	1✔
449	+ ", runId="
450	+ request.getUpdateRef().getWorkflowExecution().getRunId()	1✔
451	+ ", updateId="
452	+ request.getUpdateRef().getUpdateId()));	1✔
453	}
454	} else if (e != null) {	1✔
455	resultCF.completeExceptionally(e);	1✔
456	} else {
457	resultCF.complete(r);	1✔
458	}
459	});	1✔
460	}	1✔
461
462	@Override
463	public CancelOutput cancel(CancelInput input) {
464	RequestCancelWorkflowExecutionRequest.Builder request =
465	RequestCancelWorkflowExecutionRequest.newBuilder()	1✔
466	.setRequestId(UUID.randomUUID().toString())	1✔
467	.setWorkflowExecution(input.getWorkflowExecution())	1✔
468	.setNamespace(clientOptions.getNamespace())	1✔
469	.setIdentity(clientOptions.getIdentity());	1✔
470	genericClient.requestCancel(request.build());	1✔
471	return new CancelOutput();	1✔
472	}
473
474	@Override
475	public TerminateOutput terminate(TerminateInput input) {
476	TerminateWorkflowExecutionRequest.Builder request =
477	TerminateWorkflowExecutionRequest.newBuilder()	1✔
478	.setNamespace(clientOptions.getNamespace())	1✔
479	.setWorkflowExecution(input.getWorkflowExecution());	1✔
480	if (input.getReason() != null) {	1✔
481	request.setReason(input.getReason());	1✔
482	}
483	DataConverter dataConverterWithWorkflowContext =	1✔
484	clientOptions
485	.getDataConverter()	1✔
486	.withContext(	1✔
487	new WorkflowSerializationContext(
488	clientOptions.getNamespace(), input.getWorkflowExecution().getWorkflowId()));	1✔
489	Optional<Payloads> payloads = dataConverterWithWorkflowContext.toPayloads(input.getDetails());	1✔
490	payloads.ifPresent(request::setDetails);	1✔
491	genericClient.terminate(request.build());	1✔
492	return new TerminateOutput();	1✔
493	}
494
495	private static <R> R convertResultPayloads(
496	Optional<Payloads> resultValue,
497	Class<R> resultClass,
498	Type resultType,
499	DataConverter dataConverter) {
500	return dataConverter.fromPayloads(0, resultValue, resultClass, resultType);	1✔
501	}
502
503	/**
504	* @return a handle to dispatch the eager workflow task. {@code null} if an eager execution is
505	* disabled through {@link io.temporal.client.WorkflowOptions} or the worker
506	* <ul>
507	* <li>is activity only worker
508	* <li>not started, shutdown or paused
509	* <li>doesn't have an executor slot available
510	* </ul>
511	*/
512	@Nullable
513	private WorkflowTaskDispatchHandle obtainDispatchHandle(WorkflowStartInput input) {
514	if (input.getOptions().isDisableEagerExecution()) {	1✔
515	return null;	1✔
516	}
517	return eagerWorkflowTaskDispatcher.tryGetLocalDispatchHandler(input);	1✔
518	}
519	}

temporalio / sdk-java / #175

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