2272

Committed 23 Apr 2024 07:08PM UTC coverage: 60.17% (+0.05%) from 60.125%

Build # 2272

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Changed WorkerOption to fallback and use WorkflowClient tracer instead of NoopTracer (#883)

What changed?

WorkflowOptions will now fallback and use WorkflowClient's tracer.
Expose getOptions on IWorkflowService interface.
Why?

Make it easier to set tracer. Previously, it's required to set tracer both on the client and worker. Now setting tracer only on client should just work.

How did you test it?

Unit Test

Run Details

8 of 14 new or added lines in 9 files covered. (57.14%)

8 existing lines in 2 files now uncovered.

11472 of 19066 relevant lines covered (60.17%)

0.6 hits per line

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

72.12

/src/main/java/com/uber/cadence/migration/MigrationIWorkflowService.java

/*
 *  Copyright 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 file except in compliance with the License. A copy of the License is
 *  located at
 *
 *  http://aws.amazon.com/apache2.0
 *
 *  or in the "license" file accompanying this file. This file 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 com.uber.cadence.migration;

import com.google.common.base.Strings;
import com.uber.cadence.*;
import com.uber.cadence.serviceclient.ClientOptions;
import com.uber.cadence.serviceclient.IWorkflowService;
import com.uber.cadence.serviceclient.IWorkflowServiceBase;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import org.apache.thrift.TException;

public class MigrationIWorkflowService extends IWorkflowServiceBase {

  private IWorkflowService serviceOld, serviceNew;
  private String domainOld, domainNew;
  private static final int _defaultPageSize = 10;
  private static final String _listWorkflow = "_listWorkflow";
  private static final String _scanWorkflow = "_scanWorkflow";
  byte[] _marker = "to".getBytes();

  public MigrationIWorkflowService(
      IWorkflowService serviceOld,
      String domainOld,
      IWorkflowService serviceNew,
      String domainNew) {
    this.serviceOld = serviceOld;
    this.domainOld = domainOld;
    this.serviceNew = serviceNew;
    this.domainNew = domainNew;
  }

  @Override
  public ClientOptions getOptions() {
    return serviceOld.getOptions();
  }

  @Override
  public StartWorkflowExecutionResponse StartWorkflowExecution(
      StartWorkflowExecutionRequest startRequest) throws TException {

    if (shouldStartInNew(startRequest.getWorkflowId()))
      return serviceNew.StartWorkflowExecution(startRequest);

    return serviceOld.StartWorkflowExecution(startRequest);
  }

  @Override
  public StartWorkflowExecutionAsyncResponse StartWorkflowExecutionAsync(
      StartWorkflowExecutionAsyncRequest startRequest)
      throws BadRequestError, WorkflowExecutionAlreadyStartedError, ServiceBusyError,
          DomainNotActiveError, LimitExceededError, EntityNotExistsError,
          ClientVersionNotSupportedError, TException {

    if (shouldStartInNew(startRequest.getRequest().getWorkflowId())) {
      return serviceNew.StartWorkflowExecutionAsync(startRequest);
    }

    return serviceOld.StartWorkflowExecutionAsync(startRequest);
  }

  /**
   * SignalWithStartWorkflowExecution is used to ensure sending signal to a workflow. If the
   * workflow is running, this results in WorkflowExecutionSignaled event being recorded in the
   * history and a decision task being created for the execution. If the workflow is not running or
   * not found, this results in WorkflowExecutionStarted and WorkflowExecutionSignaled events being
   * recorded in history, and a decision task being created for the execution
   */
  @Override
  public StartWorkflowExecutionResponse SignalWithStartWorkflowExecution(
      SignalWithStartWorkflowExecutionRequest signalWithStartRequest) throws TException {
    if (shouldStartInNew(signalWithStartRequest.getWorkflowId()))
      return serviceNew.SignalWithStartWorkflowExecution(signalWithStartRequest);
    return serviceOld.SignalWithStartWorkflowExecution(signalWithStartRequest);
  }

  @Override
  public SignalWithStartWorkflowExecutionAsyncResponse SignalWithStartWorkflowExecutionAsync(
      SignalWithStartWorkflowExecutionAsyncRequest signalWithStartRequest)
      throws BadRequestError, WorkflowExecutionAlreadyStartedError, ServiceBusyError,
          DomainNotActiveError, LimitExceededError, EntityNotExistsError,
          ClientVersionNotSupportedError, TException {
    if (shouldStartInNew(signalWithStartRequest.getRequest().getWorkflowId())) {
      return serviceNew.SignalWithStartWorkflowExecutionAsync(signalWithStartRequest);
    }

    return serviceOld.SignalWithStartWorkflowExecutionAsync(signalWithStartRequest);
  }

  /**
   * SignalWorkflowExecution is used to send a signal event to running workflow execution. This
   * results in WorkflowExecutionSignaled event recorded in the history and a decision task being
   * created for the execution.
   */
  @Override
  public void SignalWorkflowExecution(SignalWorkflowExecutionRequest signalRequest)
      throws TException {
    if (shouldStartInNew(signalRequest.getWorkflowExecution().getWorkflowId()))
      serviceNew.SignalWorkflowExecution(signalRequest);
    else serviceOld.SignalWorkflowExecution(signalRequest);
  }

  @Override
  public RestartWorkflowExecutionResponse RestartWorkflowExecution(
      RestartWorkflowExecutionRequest restartRequest)
      throws BadRequestError, ServiceBusyError, DomainNotActiveError, LimitExceededError,
          EntityNotExistsError, ClientVersionNotSupportedError, TException {
    if (shouldStartInNew(restartRequest.getWorkflowExecution().getWorkflowId())) {
      return serviceNew.RestartWorkflowExecution(restartRequest);
    }

    return serviceOld.RestartWorkflowExecution(restartRequest);
  }

  @Override
  public GetWorkflowExecutionHistoryResponse GetWorkflowExecutionHistory(
      GetWorkflowExecutionHistoryRequest getRequest) throws TException {
    if (shouldStartInNew(getRequest.execution.getWorkflowId()))
      return serviceNew.GetWorkflowExecutionHistory(getRequest);
    return serviceOld.GetWorkflowExecutionHistory(getRequest);
  }

  private ListWorkflowExecutionsResponse callOldCluster(
      ListWorkflowExecutionsRequest listWorkflowExecutionsRequest,
      int pageSizeOverride,
      String searchType)
      throws TException {

    if (pageSizeOverride != 0) {
      listWorkflowExecutionsRequest.setPageSize(pageSizeOverride);
    }
    ListWorkflowExecutionsResponse response = new ListWorkflowExecutionsResponse();
    if (searchType.equals(_listWorkflow)) {
      response = serviceOld.ListWorkflowExecutions(listWorkflowExecutionsRequest);
    } else if (searchType.equals(_scanWorkflow)) {
      response = serviceOld.ScanWorkflowExecutions(listWorkflowExecutionsRequest);
    }
    return response;
  }

  private ListWorkflowExecutionsResponse appendResultsFromOldCluster(
      ListWorkflowExecutionsRequest listWorkflowExecutionsRequest,
      ListWorkflowExecutionsResponse response,
      String searchType)
      throws TException {
    int responsePageSize = response.getExecutions().size();
    int neededPageSize = listWorkflowExecutionsRequest.getPageSize() - responsePageSize;

    ListWorkflowExecutionsResponse fromResponse =
        callOldCluster(listWorkflowExecutionsRequest, neededPageSize, searchType);

    // if old cluster is empty
    if (fromResponse == null) {
      return response;
    }

    fromResponse.getExecutions().addAll(response.getExecutions());
    return fromResponse;
  }

  public boolean hasPrefix(byte[] s, byte[] prefix) {
    return s == null
        ? false
        : s.length >= prefix.length
            && Arrays.equals(Arrays.copyOfRange(s, 0, prefix.length), prefix);
  }

  /**
   * This method handles pagination and combines results from both the new and old workflow service
   * clusters. The method first checks if the nextPageToken is not set or starts with the marker
   * (_marker) to determine if it should query the new cluster (serviceNew) or combine results from
   * both the new and old clusters. If nextPageToken is set and doesn't start with the marker, it
   * queries the old cluster (serviceOld). In case the response from the new cluster is null, it
   * retries the request on the old cluster. If the number of workflow executions returned by the
   * new cluster is less than the pageSize, it appends results from the old cluster to the response.
   *
   * @param listRequest The ListWorkflowExecutionsRequest containing the query parameters, including
   *     domain, nextPageToken, pageSize, and other filtering options.
   * @return The ListWorkflowExecutionsResponse containing a list of WorkflowExecutionInfo
   *     representing the workflow executions that match the query criteria. The response also
   *     includes a nextPageToken to support pagination.
   * @throws TException if there's any communication error with the underlying workflow service.
   * @throws BadRequestError if the provided ListWorkflowExecutionsRequest is invalid (null or lacks
   *     a domain).
   */
  @Override
  public ListWorkflowExecutionsResponse ListWorkflowExecutions(
      ListWorkflowExecutionsRequest listRequest) throws TException {

    if (listRequest == null) {
      throw new BadRequestError("List request is null");
    } else if (Strings.isNullOrEmpty(listRequest.getDomain())) {
      throw new BadRequestError("Domain is null or empty");
    }
    if (!listRequest.isSetPageSize()) {
      listRequest.pageSize = _defaultPageSize;
    }

    if (!listRequest.isSetNextPageToken()
        || listRequest.getNextPageToken().length == 0
        || hasPrefix(listRequest.getNextPageToken(), _marker)) {
      if (hasPrefix(listRequest.getNextPageToken(), _marker) == true) {
        listRequest.setNextPageToken(
            Arrays.copyOfRange(
                listRequest.getNextPageToken(),
                _marker.length,
                listRequest.getNextPageToken().length));
      }
      ListWorkflowExecutionsResponse response = serviceNew.ListWorkflowExecutions(listRequest);
      if (response == null) return callOldCluster(listRequest, 0, _listWorkflow);

      if (response.getExecutions().size() < listRequest.getPageSize()) {
        return appendResultsFromOldCluster(listRequest, response, _listWorkflow);
      }

      byte[] combinedNextPageToken = new byte[_marker.length + response.getNextPageToken().length];
      System.arraycopy(_marker, 0, combinedNextPageToken, 0, _marker.length);
      System.arraycopy(
          response.getNextPageToken(),
          0,
          combinedNextPageToken,
          _marker.length,
          response.getNextPageToken().length);
      response.setNextPageToken(combinedNextPageToken);
      return response;
    }
    return callOldCluster(listRequest, 0, _listWorkflow);
  }

  /**
   * Scans workflow executions based on the provided request parameters, handling pagination and
   * combining results from the new and old clusters. The method queries the new cluster
   * (serviceNew) if nextPageToken is not set or starts with the marker (_marker). Otherwise, it
   * queries the old cluster (serviceOld). Results from the old cluster are appended if needed to
   * maintain correct pagination.
   *
   * @param listRequest The ListWorkflowExecutionsRequest containing query parameters.
   * @return The ListWorkflowExecutionsResponse with WorkflowExecutionInfo and nextPageToken.
   * @throws TException if there's any communication error with the workflow service.
   * @throws BadRequestError if the provided ListWorkflowExecutionsRequest is invalid.
   */
  @Override
  public ListWorkflowExecutionsResponse ScanWorkflowExecutions(
      ListWorkflowExecutionsRequest listRequest) throws TException {
    ListWorkflowExecutionsResponse response;
    if (listRequest == null) {
      throw new BadRequestError("List request is null");
    } else if (Strings.isNullOrEmpty(listRequest.getDomain())) {
      throw new BadRequestError("Domain is null or empty");
    }
    if (!listRequest.isSetPageSize()) {
      listRequest.pageSize = _defaultPageSize;
    }

    if (!listRequest.isSetNextPageToken()
        || listRequest.getNextPageToken().length == 0
        || hasPrefix(listRequest.getNextPageToken(), _marker)) {
      if (hasPrefix(listRequest.getNextPageToken(), _marker)) {
        listRequest.setNextPageToken(
            Arrays.copyOfRange(
                listRequest.getNextPageToken(),
                _marker.length,
                listRequest.getNextPageToken().length));
      }
      response = serviceNew.ScanWorkflowExecutions(listRequest);
      if (response == null) return callOldCluster(listRequest, 0, _scanWorkflow);

      if (response.getExecutions().size() < listRequest.getPageSize()) {
        return appendResultsFromOldCluster(listRequest, response, _scanWorkflow);
      }

      byte[] combinedNextPageToken = new byte[_marker.length + response.getNextPageToken().length];
      System.arraycopy(_marker, 0, combinedNextPageToken, 0, _marker.length);
      System.arraycopy(
          response.getNextPageToken(),
          0,
          combinedNextPageToken,
          _marker.length,
          response.getNextPageToken().length);
      response.setNextPageToken(combinedNextPageToken);
      return response;
    }
    return callOldCluster(listRequest, 0, _scanWorkflow);
  }

  @Override
  public ListOpenWorkflowExecutionsResponse ListOpenWorkflowExecutions(
      ListOpenWorkflowExecutionsRequest listRequest) throws TException {
    ListOpenWorkflowExecutionsResponse response;
    if (listRequest == null) {
      throw new BadRequestError("List request is null");
    } else if (Strings.isNullOrEmpty(listRequest.getDomain())) {
      throw new BadRequestError("Domain is null or empty");
    }
    if (!listRequest.isSetMaximumPageSize()) {
      listRequest.maximumPageSize = _defaultPageSize;
    }

    if (!listRequest.isSetNextPageToken()
        || listRequest.getNextPageToken().length == 0
        || hasPrefix(listRequest.getNextPageToken(), _marker)) {
      if (hasPrefix(listRequest.getNextPageToken(), _marker)) {
        listRequest.setNextPageToken(
            Arrays.copyOfRange(
                listRequest.getNextPageToken(),
                _marker.length,
                listRequest.getNextPageToken().length));
      }
      response = serviceNew.ListOpenWorkflowExecutions(listRequest);
      if (response == null) return serviceOld.ListOpenWorkflowExecutions(listRequest);

      if (response.getExecutionsSize() < listRequest.getMaximumPageSize()) {
        int neededPageSize = listRequest.getMaximumPageSize() - response.getExecutionsSize();
        ListOpenWorkflowExecutionsRequest copiedRequest =
            new ListOpenWorkflowExecutionsRequest(listRequest);
        copiedRequest.maximumPageSize = neededPageSize;
        ListOpenWorkflowExecutionsResponse fromResponse =
            serviceOld.ListOpenWorkflowExecutions(copiedRequest);
        if (fromResponse == null) return response;

        fromResponse.getExecutions().addAll(response.getExecutions());
        return fromResponse;
      }

      byte[] combinedNextPageToken = new byte[_marker.length + response.getNextPageToken().length];
      System.arraycopy(_marker, 0, combinedNextPageToken, 0, _marker.length);
      System.arraycopy(
          response.getNextPageToken(),
          0,
          combinedNextPageToken,
          _marker.length,
          response.getNextPageToken().length);
      response.setNextPageToken(combinedNextPageToken);
      return response;
    }
    return serviceOld.ListOpenWorkflowExecutions(listRequest);
  }

  @Override
  public ListClosedWorkflowExecutionsResponse ListClosedWorkflowExecutions(
      ListClosedWorkflowExecutionsRequest listRequest) throws TException {
    ListClosedWorkflowExecutionsResponse response;
    if (listRequest == null) {
      throw new BadRequestError("List request is null");
    } else if (Strings.isNullOrEmpty(listRequest.getDomain())) {
      throw new BadRequestError("Domain is null or empty");
    }
    if (!listRequest.isSetMaximumPageSize()) {
      listRequest.maximumPageSize = _defaultPageSize;
    }

    if (!listRequest.isSetNextPageToken()
        || listRequest.getNextPageToken().length == 0
        || hasPrefix(listRequest.getNextPageToken(), _marker)) {
      if (hasPrefix(listRequest.getNextPageToken(), _marker)) {
        listRequest.setNextPageToken(
            Arrays.copyOfRange(
                listRequest.getNextPageToken(),
                _marker.length,
                listRequest.getNextPageToken().length));
      }
      response = serviceNew.ListClosedWorkflowExecutions(listRequest);
      if (response == null) return serviceOld.ListClosedWorkflowExecutions(listRequest);

      if (response.getExecutionsSize() < listRequest.getMaximumPageSize()) {
        int neededPageSize = listRequest.getMaximumPageSize() - response.getExecutionsSize();
        ListClosedWorkflowExecutionsRequest copiedRequest =
            new ListClosedWorkflowExecutionsRequest(listRequest);
        copiedRequest.maximumPageSize = neededPageSize;
        ListClosedWorkflowExecutionsResponse fromResponse =
            serviceOld.ListClosedWorkflowExecutions(copiedRequest);
        if (fromResponse == null) return response;

        fromResponse.getExecutions().addAll(response.getExecutions());
        return fromResponse;
      }

      byte[] combinedNextPageToken = new byte[_marker.length + response.getNextPageToken().length];
      System.arraycopy(_marker, 0, combinedNextPageToken, 0, _marker.length);
      System.arraycopy(
          response.getNextPageToken(),
          0,
          combinedNextPageToken,
          _marker.length,
          response.getNextPageToken().length);
      response.setNextPageToken(combinedNextPageToken);
      return response;
    }
    return serviceOld.ListClosedWorkflowExecutions(listRequest);
  }

  @Override
  public QueryWorkflowResponse QueryWorkflow(QueryWorkflowRequest queryRequest) throws TException {

    try {
      if (shouldStartInNew(queryRequest.getExecution().getWorkflowId()))
        return serviceNew.QueryWorkflow(queryRequest);
      return serviceOld.QueryWorkflow(queryRequest);
    } catch (NullPointerException e) {
      throw new NullPointerException(
          "Query does not have workflowID associated: " + e.getMessage());
    }
  }

  @Override
  public CountWorkflowExecutionsResponse CountWorkflowExecutions(
      CountWorkflowExecutionsRequest countRequest) throws TException {

    CountWorkflowExecutionsResponse countResponseNew =
        serviceNew.CountWorkflowExecutions(countRequest);
    CountWorkflowExecutionsResponse countResponseOld =
        serviceOld.CountWorkflowExecutions(countRequest);
    if (countResponseNew == null) return countResponseOld;
    if (countResponseOld == null) return countResponseNew;

    countResponseOld.setCount(countResponseOld.getCount() + countResponseNew.getCount());
    return countResponseOld;
  }

  @Override
  public void TerminateWorkflowExecution(TerminateWorkflowExecutionRequest terminateRequest)
      throws TException {
    try {
      serviceNew.TerminateWorkflowExecution(terminateRequest);
    } catch (EntityNotExistsError e) {
      serviceOld.TerminateWorkflowExecution(terminateRequest);
    }
  }

  private Boolean shouldStartInNew(String workflowID) throws TException {
    try {
      return describeWorkflowExecution(serviceNew, domainNew, workflowID)
          .thenCombine(
              describeWorkflowExecution(serviceOld, domainOld, workflowID),
              (respNew, respOld) ->
                  respNew != null // execution already in new
                      || respOld == null // execution not exist in new and not exist in old
                      || (respOld.isSetWorkflowExecutionInfo()
                          && respOld
                              .getWorkflowExecutionInfo()
                              .isSetCloseStatus()) // execution not exist in new and execution is
              // closed in old
              )
          .get();
    } catch (CompletionException e) {
      throw e.getCause() instanceof TException
          ? (TException) e.getCause()
          : new TException("unknown error: " + e.getMessage());
    } catch (Exception e) {
      throw new TException("Unknown error: " + e.getMessage());
    }
  }

  private CompletableFuture<DescribeWorkflowExecutionResponse> describeWorkflowExecution(
      IWorkflowService service, String domain, String workflowID) {
    return CompletableFuture.supplyAsync(
        () -> {
          try {
            return service.DescribeWorkflowExecution(
                new DescribeWorkflowExecutionRequest()
                    .setDomain(domain)
                    .setExecution(new WorkflowExecution().setWorkflowId(workflowID)));
          } catch (EntityNotExistsError e) {
            return null;
          } catch (Exception e) {
            throw new CompletionException(e);
          }
        });
  }
}

1	/*
2	* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3	*
4	* Modifications copyright (C) 2017 Uber Technologies, Inc.
5	*
6	* Licensed under the Apache License, Version 2.0 (the "License"). You may not
7	* use this file except in compliance with the License. A copy of the License is
8	* located at
9	*
10	* http://aws.amazon.com/apache2.0
11	*
12	* or in the "license" file accompanying this file. This file is distributed on
13	* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14	* express or implied. See the License for the specific language governing
15	* permissions and limitations under the License.
16	*/
17
18	package com.uber.cadence.migration;
19
20	import com.google.common.base.Strings;
21	import com.uber.cadence.*;
22	import com.uber.cadence.serviceclient.ClientOptions;
23	import com.uber.cadence.serviceclient.IWorkflowService;
24	import com.uber.cadence.serviceclient.IWorkflowServiceBase;
25	import java.util.Arrays;
26	import java.util.concurrent.CompletableFuture;
27	import java.util.concurrent.CompletionException;
28	import org.apache.thrift.TException;
29
30	public class MigrationIWorkflowService extends IWorkflowServiceBase {
31
32	private IWorkflowService serviceOld, serviceNew;
33	private String domainOld, domainNew;
34	private static final int _defaultPageSize = 10;
35	private static final String _listWorkflow = "_listWorkflow";
36	private static final String _scanWorkflow = "_scanWorkflow";
37	byte[] _marker = "to".getBytes();	1✔
38
39	public MigrationIWorkflowService(
40	IWorkflowService serviceOld,
41	String domainOld,
42	IWorkflowService serviceNew,
43	String domainNew) {	1✔
44	this.serviceOld = serviceOld;	1✔
45	this.domainOld = domainOld;	1✔
46	this.serviceNew = serviceNew;	1✔
47	this.domainNew = domainNew;	1✔
48	}	1✔
49
50	@Override
51	public ClientOptions getOptions() {
NEW 52	return serviceOld.getOptions();	×
53	}
54
55	@Override
56	public StartWorkflowExecutionResponse StartWorkflowExecution(
57	StartWorkflowExecutionRequest startRequest) throws TException {
58
59	if (shouldStartInNew(startRequest.getWorkflowId()))	1✔
60	return serviceNew.StartWorkflowExecution(startRequest);	1✔
61
62	return serviceOld.StartWorkflowExecution(startRequest);	1✔
63	}
64
65	@Override
66	public StartWorkflowExecutionAsyncResponse StartWorkflowExecutionAsync(
67	StartWorkflowExecutionAsyncRequest startRequest)
68	throws BadRequestError, WorkflowExecutionAlreadyStartedError, ServiceBusyError,
69	DomainNotActiveError, LimitExceededError, EntityNotExistsError,
70	ClientVersionNotSupportedError, TException {
71
72	if (shouldStartInNew(startRequest.getRequest().getWorkflowId())) {	×
73	return serviceNew.StartWorkflowExecutionAsync(startRequest);	×
74	}
75
76	return serviceOld.StartWorkflowExecutionAsync(startRequest);	×
77	}
78
79	/**
80	* SignalWithStartWorkflowExecution is used to ensure sending signal to a workflow. If the
81	* workflow is running, this results in WorkflowExecutionSignaled event being recorded in the
82	* history and a decision task being created for the execution. If the workflow is not running or
83	* not found, this results in WorkflowExecutionStarted and WorkflowExecutionSignaled events being
84	* recorded in history, and a decision task being created for the execution
85	*/
86	@Override
87	public StartWorkflowExecutionResponse SignalWithStartWorkflowExecution(
88	SignalWithStartWorkflowExecutionRequest signalWithStartRequest) throws TException {
89	if (shouldStartInNew(signalWithStartRequest.getWorkflowId()))	1✔
90	return serviceNew.SignalWithStartWorkflowExecution(signalWithStartRequest);	1✔
91	return serviceOld.SignalWithStartWorkflowExecution(signalWithStartRequest);	1✔
92	}
93
94	@Override
95	public SignalWithStartWorkflowExecutionAsyncResponse SignalWithStartWorkflowExecutionAsync(
96	SignalWithStartWorkflowExecutionAsyncRequest signalWithStartRequest)
97	throws BadRequestError, WorkflowExecutionAlreadyStartedError, ServiceBusyError,
98	DomainNotActiveError, LimitExceededError, EntityNotExistsError,
99	ClientVersionNotSupportedError, TException {
100	if (shouldStartInNew(signalWithStartRequest.getRequest().getWorkflowId())) {	×
101	return serviceNew.SignalWithStartWorkflowExecutionAsync(signalWithStartRequest);	×
102	}
103
104	return serviceOld.SignalWithStartWorkflowExecutionAsync(signalWithStartRequest);	×
105	}
106
107	/**
108	* SignalWorkflowExecution is used to send a signal event to running workflow execution. This
109	* results in WorkflowExecutionSignaled event recorded in the history and a decision task being
110	* created for the execution.
111	*/
112	@Override
113	public void SignalWorkflowExecution(SignalWorkflowExecutionRequest signalRequest)
114	throws TException {
115	if (shouldStartInNew(signalRequest.getWorkflowExecution().getWorkflowId()))	1✔
116	serviceNew.SignalWorkflowExecution(signalRequest);	1✔
117	else serviceOld.SignalWorkflowExecution(signalRequest);	1✔
118	}	1✔
119
120	@Override
121	public RestartWorkflowExecutionResponse RestartWorkflowExecution(
122	RestartWorkflowExecutionRequest restartRequest)
123	throws BadRequestError, ServiceBusyError, DomainNotActiveError, LimitExceededError,
124	EntityNotExistsError, ClientVersionNotSupportedError, TException {
125	if (shouldStartInNew(restartRequest.getWorkflowExecution().getWorkflowId())) {	×
126	return serviceNew.RestartWorkflowExecution(restartRequest);	×
127	}
128
129	return serviceOld.RestartWorkflowExecution(restartRequest);	×
130	}
131
132	@Override
133	public GetWorkflowExecutionHistoryResponse GetWorkflowExecutionHistory(
134	GetWorkflowExecutionHistoryRequest getRequest) throws TException {
135	if (shouldStartInNew(getRequest.execution.getWorkflowId()))	×
136	return serviceNew.GetWorkflowExecutionHistory(getRequest);	×
137	return serviceOld.GetWorkflowExecutionHistory(getRequest);	×
138	}
139
140	private ListWorkflowExecutionsResponse callOldCluster(
141	ListWorkflowExecutionsRequest listWorkflowExecutionsRequest,
142	int pageSizeOverride,
143	String searchType)
144	throws TException {
145
146	if (pageSizeOverride != 0) {	1✔
147	listWorkflowExecutionsRequest.setPageSize(pageSizeOverride);	1✔
148	}
149	ListWorkflowExecutionsResponse response = new ListWorkflowExecutionsResponse();	1✔
150	if (searchType.equals(_listWorkflow)) {	1✔
151	response = serviceOld.ListWorkflowExecutions(listWorkflowExecutionsRequest);	1✔
152	} else if (searchType.equals(_scanWorkflow)) {	1✔
153	response = serviceOld.ScanWorkflowExecutions(listWorkflowExecutionsRequest);	1✔
154	}
155	return response;	1✔
156	}
157
158	private ListWorkflowExecutionsResponse appendResultsFromOldCluster(
159	ListWorkflowExecutionsRequest listWorkflowExecutionsRequest,
160	ListWorkflowExecutionsResponse response,
161	String searchType)
162	throws TException {
163	int responsePageSize = response.getExecutions().size();	1✔
164	int neededPageSize = listWorkflowExecutionsRequest.getPageSize() - responsePageSize;	1✔
165
166	ListWorkflowExecutionsResponse fromResponse =	1✔
167	callOldCluster(listWorkflowExecutionsRequest, neededPageSize, searchType);	1✔
168
169	// if old cluster is empty
170	if (fromResponse == null) {	1✔
171	return response;	1✔
172	}
173
174	fromResponse.getExecutions().addAll(response.getExecutions());	1✔
175	return fromResponse;	1✔
176	}
177
178	public boolean hasPrefix(byte[] s, byte[] prefix) {
179	return s == null	1✔
180	? false	1✔
181	: s.length >= prefix.length
182	&& Arrays.equals(Arrays.copyOfRange(s, 0, prefix.length), prefix);	×
183	}
184
185	/**
186	* This method handles pagination and combines results from both the new and old workflow service
187	* clusters. The method first checks if the nextPageToken is not set or starts with the marker
188	* (_marker) to determine if it should query the new cluster (serviceNew) or combine results from
189	* both the new and old clusters. If nextPageToken is set and doesn't start with the marker, it
190	* queries the old cluster (serviceOld). In case the response from the new cluster is null, it
191	* retries the request on the old cluster. If the number of workflow executions returned by the
192	* new cluster is less than the pageSize, it appends results from the old cluster to the response.
193	*
194	* @param listRequest The ListWorkflowExecutionsRequest containing the query parameters, including
195	* domain, nextPageToken, pageSize, and other filtering options.
196	* @return The ListWorkflowExecutionsResponse containing a list of WorkflowExecutionInfo
197	* representing the workflow executions that match the query criteria. The response also
198	* includes a nextPageToken to support pagination.
199	* @throws TException if there's any communication error with the underlying workflow service.
200	* @throws BadRequestError if the provided ListWorkflowExecutionsRequest is invalid (null or lacks
201	* a domain).
202	*/
203	@Override
204	public ListWorkflowExecutionsResponse ListWorkflowExecutions(
205	ListWorkflowExecutionsRequest listRequest) throws TException {
206
207	if (listRequest == null) {	1✔
208	throw new BadRequestError("List request is null");	1✔
209	} else if (Strings.isNullOrEmpty(listRequest.getDomain())) {	1✔
210	throw new BadRequestError("Domain is null or empty");	1✔
211	}
212	if (!listRequest.isSetPageSize()) {	1✔
213	listRequest.pageSize = _defaultPageSize;	1✔
214	}
215
216	if (!listRequest.isSetNextPageToken()	1✔
217	\|\| listRequest.getNextPageToken().length == 0	×
218	\|\| hasPrefix(listRequest.getNextPageToken(), _marker)) {	×
219	if (hasPrefix(listRequest.getNextPageToken(), _marker) == true) {	1✔
220	listRequest.setNextPageToken(	×
221	Arrays.copyOfRange(	×
222	listRequest.getNextPageToken(),	×
223	_marker.length,
224	listRequest.getNextPageToken().length));	×
225	}
226	ListWorkflowExecutionsResponse response = serviceNew.ListWorkflowExecutions(listRequest);	1✔
227	if (response == null) return callOldCluster(listRequest, 0, _listWorkflow);	1✔
228
229	if (response.getExecutions().size() < listRequest.getPageSize()) {	1✔
230	return appendResultsFromOldCluster(listRequest, response, _listWorkflow);	1✔
231	}
232
233	byte[] combinedNextPageToken = new byte[_marker.length + response.getNextPageToken().length];	1✔
234	System.arraycopy(_marker, 0, combinedNextPageToken, 0, _marker.length);	1✔
235	System.arraycopy(	1✔
236	response.getNextPageToken(),	1✔
237	0,
238	combinedNextPageToken,
239	_marker.length,
240	response.getNextPageToken().length);	1✔
241	response.setNextPageToken(combinedNextPageToken);	1✔
242	return response;	1✔
243	}
244	return callOldCluster(listRequest, 0, _listWorkflow);	×
245	}
246
247	/**
248	* Scans workflow executions based on the provided request parameters, handling pagination and
249	* combining results from the new and old clusters. The method queries the new cluster
250	* (serviceNew) if nextPageToken is not set or starts with the marker (_marker). Otherwise, it
251	* queries the old cluster (serviceOld). Results from the old cluster are appended if needed to
252	* maintain correct pagination.
253	*
254	* @param listRequest The ListWorkflowExecutionsRequest containing query parameters.
255	* @return The ListWorkflowExecutionsResponse with WorkflowExecutionInfo and nextPageToken.
256	* @throws TException if there's any communication error with the workflow service.
257	* @throws BadRequestError if the provided ListWorkflowExecutionsRequest is invalid.
258	*/
259	@Override
260	public ListWorkflowExecutionsResponse ScanWorkflowExecutions(
261	ListWorkflowExecutionsRequest listRequest) throws TException {
262	ListWorkflowExecutionsResponse response;
263	if (listRequest == null) {	1✔
264	throw new BadRequestError("List request is null");	1✔
265	} else if (Strings.isNullOrEmpty(listRequest.getDomain())) {	1✔
266	throw new BadRequestError("Domain is null or empty");	1✔
267	}
268	if (!listRequest.isSetPageSize()) {	1✔
269	listRequest.pageSize = _defaultPageSize;	1✔
270	}
271
272	if (!listRequest.isSetNextPageToken()	1✔
273	\|\| listRequest.getNextPageToken().length == 0	×
274	\|\| hasPrefix(listRequest.getNextPageToken(), _marker)) {	×
275	if (hasPrefix(listRequest.getNextPageToken(), _marker)) {	1✔
276	listRequest.setNextPageToken(	×
277	Arrays.copyOfRange(	×
278	listRequest.getNextPageToken(),	×
279	_marker.length,
280	listRequest.getNextPageToken().length));	×
281	}
282	response = serviceNew.ScanWorkflowExecutions(listRequest);	1✔
283	if (response == null) return callOldCluster(listRequest, 0, _scanWorkflow);	1✔
284
285	if (response.getExecutions().size() < listRequest.getPageSize()) {	1✔
286	return appendResultsFromOldCluster(listRequest, response, _scanWorkflow);	1✔
287	}
288
289	byte[] combinedNextPageToken = new byte[_marker.length + response.getNextPageToken().length];	1✔
290	System.arraycopy(_marker, 0, combinedNextPageToken, 0, _marker.length);	1✔
291	System.arraycopy(	1✔
292	response.getNextPageToken(),	1✔
293	0,
294	combinedNextPageToken,
295	_marker.length,
296	response.getNextPageToken().length);	1✔
297	response.setNextPageToken(combinedNextPageToken);	1✔
298	return response;	1✔
299	}
300	return callOldCluster(listRequest, 0, _scanWorkflow);	×
301	}
302
303	@Override
304	public ListOpenWorkflowExecutionsResponse ListOpenWorkflowExecutions(
305	ListOpenWorkflowExecutionsRequest listRequest) throws TException {
306	ListOpenWorkflowExecutionsResponse response;
307	if (listRequest == null) {	1✔
308	throw new BadRequestError("List request is null");	1✔
309	} else if (Strings.isNullOrEmpty(listRequest.getDomain())) {	1✔
310	throw new BadRequestError("Domain is null or empty");	1✔
311	}
312	if (!listRequest.isSetMaximumPageSize()) {	1✔
313	listRequest.maximumPageSize = _defaultPageSize;	1✔
314	}
315
316	if (!listRequest.isSetNextPageToken()	1✔
317	\|\| listRequest.getNextPageToken().length == 0	×
318	\|\| hasPrefix(listRequest.getNextPageToken(), _marker)) {	×
319	if (hasPrefix(listRequest.getNextPageToken(), _marker)) {	1✔
320	listRequest.setNextPageToken(	×
321	Arrays.copyOfRange(	×
322	listRequest.getNextPageToken(),	×
323	_marker.length,
324	listRequest.getNextPageToken().length));	×
325	}
326	response = serviceNew.ListOpenWorkflowExecutions(listRequest);	1✔
327	if (response == null) return serviceOld.ListOpenWorkflowExecutions(listRequest);	1✔
328
329	if (response.getExecutionsSize() < listRequest.getMaximumPageSize()) {	1✔
330	int neededPageSize = listRequest.getMaximumPageSize() - response.getExecutionsSize();	1✔
331	ListOpenWorkflowExecutionsRequest copiedRequest =	1✔
332	new ListOpenWorkflowExecutionsRequest(listRequest);
333	copiedRequest.maximumPageSize = neededPageSize;	1✔
334	ListOpenWorkflowExecutionsResponse fromResponse =	1✔
335	serviceOld.ListOpenWorkflowExecutions(copiedRequest);	1✔
336	if (fromResponse == null) return response;	1✔
337
338	fromResponse.getExecutions().addAll(response.getExecutions());	1✔
339	return fromResponse;	1✔
340	}
341
342	byte[] combinedNextPageToken = new byte[_marker.length + response.getNextPageToken().length];	1✔
343	System.arraycopy(_marker, 0, combinedNextPageToken, 0, _marker.length);	1✔
344	System.arraycopy(	1✔
345	response.getNextPageToken(),	1✔
346	0,
347	combinedNextPageToken,
348	_marker.length,
349	response.getNextPageToken().length);	1✔
350	response.setNextPageToken(combinedNextPageToken);	1✔
351	return response;	1✔
352	}
353	return serviceOld.ListOpenWorkflowExecutions(listRequest);	×
354	}
355
356	@Override
357	public ListClosedWorkflowExecutionsResponse ListClosedWorkflowExecutions(
358	ListClosedWorkflowExecutionsRequest listRequest) throws TException {
359	ListClosedWorkflowExecutionsResponse response;
360	if (listRequest == null) {	1✔
361	throw new BadRequestError("List request is null");	1✔
362	} else if (Strings.isNullOrEmpty(listRequest.getDomain())) {	1✔
363	throw new BadRequestError("Domain is null or empty");	1✔
364	}
365	if (!listRequest.isSetMaximumPageSize()) {	1✔
366	listRequest.maximumPageSize = _defaultPageSize;	1✔
367	}
368
369	if (!listRequest.isSetNextPageToken()	1✔
370	\|\| listRequest.getNextPageToken().length == 0	×
371	\|\| hasPrefix(listRequest.getNextPageToken(), _marker)) {	×
372	if (hasPrefix(listRequest.getNextPageToken(), _marker)) {	1✔
373	listRequest.setNextPageToken(	×
374	Arrays.copyOfRange(	×
375	listRequest.getNextPageToken(),	×
376	_marker.length,
377	listRequest.getNextPageToken().length));	×
378	}
379	response = serviceNew.ListClosedWorkflowExecutions(listRequest);	1✔
380	if (response == null) return serviceOld.ListClosedWorkflowExecutions(listRequest);	1✔
381
382	if (response.getExecutionsSize() < listRequest.getMaximumPageSize()) {	1✔
383	int neededPageSize = listRequest.getMaximumPageSize() - response.getExecutionsSize();	1✔
384	ListClosedWorkflowExecutionsRequest copiedRequest =	1✔
385	new ListClosedWorkflowExecutionsRequest(listRequest);
386	copiedRequest.maximumPageSize = neededPageSize;	1✔
387	ListClosedWorkflowExecutionsResponse fromResponse =	1✔
388	serviceOld.ListClosedWorkflowExecutions(copiedRequest);	1✔
389	if (fromResponse == null) return response;	1✔
390
391	fromResponse.getExecutions().addAll(response.getExecutions());	1✔
392	return fromResponse;	1✔
393	}
394
395	byte[] combinedNextPageToken = new byte[_marker.length + response.getNextPageToken().length];	1✔
396	System.arraycopy(_marker, 0, combinedNextPageToken, 0, _marker.length);	1✔
397	System.arraycopy(	1✔
398	response.getNextPageToken(),	1✔
399	0,
400	combinedNextPageToken,
401	_marker.length,
402	response.getNextPageToken().length);	1✔
403	response.setNextPageToken(combinedNextPageToken);	1✔
404	return response;	1✔
405	}
406	return serviceOld.ListClosedWorkflowExecutions(listRequest);	×
407	}
408
409	@Override
410	public QueryWorkflowResponse QueryWorkflow(QueryWorkflowRequest queryRequest) throws TException {
411
412	try {
413	if (shouldStartInNew(queryRequest.getExecution().getWorkflowId()))	1✔
414	return serviceNew.QueryWorkflow(queryRequest);	1✔
415	return serviceOld.QueryWorkflow(queryRequest);	1✔
416	} catch (NullPointerException e) {	1✔
417	throw new NullPointerException(	1✔
418	"Query does not have workflowID associated: " + e.getMessage());	1✔
419	}
420	}
421
422	@Override
423	public CountWorkflowExecutionsResponse CountWorkflowExecutions(
424	CountWorkflowExecutionsRequest countRequest) throws TException {
425
426	CountWorkflowExecutionsResponse countResponseNew =	1✔
427	serviceNew.CountWorkflowExecutions(countRequest);	1✔
428	CountWorkflowExecutionsResponse countResponseOld =	1✔
429	serviceOld.CountWorkflowExecutions(countRequest);	1✔
430	if (countResponseNew == null) return countResponseOld;	1✔
431	if (countResponseOld == null) return countResponseNew;	1✔
432
433	countResponseOld.setCount(countResponseOld.getCount() + countResponseNew.getCount());	1✔
434	return countResponseOld;	1✔
435	}
436
437	@Override
438	public void TerminateWorkflowExecution(TerminateWorkflowExecutionRequest terminateRequest)
439	throws TException {
440	try {
441	serviceNew.TerminateWorkflowExecution(terminateRequest);	×
442	} catch (EntityNotExistsError e) {	×
443	serviceOld.TerminateWorkflowExecution(terminateRequest);	×
444	}	×
445	}	×
446
447	private Boolean shouldStartInNew(String workflowID) throws TException {
448	try {
449	return describeWorkflowExecution(serviceNew, domainNew, workflowID)	1✔
450	.thenCombine(	1✔
451	describeWorkflowExecution(serviceOld, domainOld, workflowID),	1✔
452	(respNew, respOld) ->
453	respNew != null // execution already in new	1✔
454	\|\| respOld == null // execution not exist in new and not exist in old
455	\|\| (respOld.isSetWorkflowExecutionInfo()	1✔
456	&& respOld
457	.getWorkflowExecutionInfo()	×
458	.isSetCloseStatus()) // execution not exist in new and execution is	1✔
459	// closed in old
460	)
461	.get();	1✔
462	} catch (CompletionException e) {	×
463	throw e.getCause() instanceof TException	×
464	? (TException) e.getCause()	×
465	: new TException("unknown error: " + e.getMessage());	×
466	} catch (Exception e) {	×
467	throw new TException("Unknown error: " + e.getMessage());	×
468	}
469	}
470
471	private CompletableFuture<DescribeWorkflowExecutionResponse> describeWorkflowExecution(
472	IWorkflowService service, String domain, String workflowID) {
473	return CompletableFuture.supplyAsync(	1✔
474	() -> {
475	try {
476	return service.DescribeWorkflowExecution(	1✔
477	new DescribeWorkflowExecutionRequest()
478	.setDomain(domain)	1✔
479	.setExecution(new WorkflowExecution().setWorkflowId(workflowID)));	1✔
480	} catch (EntityNotExistsError e) {	×
481	return null;	×
482	} catch (Exception e) {	×
483	throw new CompletionException(e);	×
484	}
485	});
486	}
487	}

uber / cadence-java-client / 2272

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