#188

Committed 25 Sep 2023 04:42PM UTC coverage: 77.369% (-0.3%) from 77.663%

Build # #188

Build Type

push

github-actions

Committed by

web-flow

Commit Message

Fix null pointer on trigger immediately (#1865)

Run Details

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

18670 of 24131 relevant lines covered (77.37%)

0.77 hits per line

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

93.17

/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.internal.common.HeaderUtils;
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())
            .setHeader(HeaderUtils.toHeaderGrpc(input.getHeader(), null));

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

temporalio / sdk-java / #188

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