2555

Committed 24 Oct 2024 10:50PM UTC coverage: 66.622% (+0.4%) from 66.195%

Build # 2555

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Refactor Test environment initialization to CadenceTestRule from WorkflowTest. (#923)

WorkflowTest is currently 6,000 lines long and has nearly every test related to end to end client behavior. It provides the rather neat behavior that it supports running against both an instance of Cadence running in Docker and against the test version. It's additionally parameterized to run the entire test suite with or without sticky execution enabled.

Due to the complexity in handling both environments, adding yet another test to WorkflowTest has always been the easiest option for developers. To allow for tests to easily be split into other files, extract the core functionality to a Junit test rule that can easily be reused by additional tests.

With the exception of testSignalCrossDomainExternalWorkflow and the replay tests that don't use the test environment, all tests have been left in WorkflowTest to be split out later.

Run Details

12910 of 19378 relevant lines covered (66.62%)

0.67 hits per line

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

86.44

/src/main/java/com/uber/cadence/internal/sync/WorkflowRetryerInternal.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.internal.sync;

import com.uber.cadence.common.RetryOptions;
import com.uber.cadence.workflow.ActivityFailureException;
import com.uber.cadence.workflow.CompletablePromise;
import com.uber.cadence.workflow.Functions;
import com.uber.cadence.workflow.Promise;
import com.uber.cadence.workflow.Workflow;
import java.time.Duration;

/**
 * Implements operation retry logic for both synchronous and asynchronous operations. Internal
 * class. Do not reference this class directly. Use {@link Workflow#retry(RetryOptions,
 * Functions.Func)} or Async{@link #retry(RetryOptions, Functions.Func)}.
 */
final class WorkflowRetryerInternal {

  public static <R> R validateOptionsAndRetry(RetryOptions options, Functions.Func<R> func) {
    return retry(RetryOptions.merge(null, options), func);
  }

  /**
   * Retry function synchronously.
   *
   * @param options retry options.
   * @param func procedure to retry.
   * @return result of func if ever completed successfully.
   */
  public static <R> R retry(RetryOptions options, Functions.Func<R> func) {
    options.validate();
    int attempt = 1;
    long startTime = WorkflowInternal.currentTimeMillis();
    // Records retry options in the history allowing changing them without breaking determinism.
    String retryId = WorkflowInternal.randomUUID().toString();
    RetryOptions retryOptions =
        WorkflowInternal.mutableSideEffect(
            retryId, RetryOptions.class, RetryOptions.class, Object::equals, () -> options);
    while (true) {
      long nextSleepTime = retryOptions.calculateSleepTime(attempt);
      try {
        return func.apply();
      } catch (Exception e) {
        long elapsed = WorkflowInternal.currentTimeMillis() - startTime;
        if (retryOptions.shouldRethrow(e, attempt, elapsed, nextSleepTime)) {
          throw WorkflowInternal.wrap(e);
        }
      }
      attempt++;
      WorkflowInternal.sleep(Duration.ofMillis(nextSleepTime));
    }
  }

  /**
   * Retry function asynchronously.
   *
   * @param options retry options.
   * @param func procedure to retry.
   * @return result promise to the result or failure if retries stopped according to options.
   */
  public static <R> Promise<R> retryAsync(RetryOptions options, Functions.Func<Promise<R>> func) {
    String retryId = WorkflowInternal.randomUUID().toString();
    long startTime = WorkflowInternal.currentTimeMillis();
    return retryAsync(retryId, options, func, startTime, 1);
  }

  private static <R> Promise<R> retryAsync(
      String retryId,
      RetryOptions options,
      Functions.Func<Promise<R>> func,
      long startTime,
      long attempt) {
    options.validate();
    RetryOptions retryOptions =
        WorkflowInternal.mutableSideEffect(
            retryId, RetryOptions.class, RetryOptions.class, Object::equals, () -> options);

    CompletablePromise<R> funcResult = WorkflowInternal.newCompletablePromise();
    try {
      funcResult.completeFrom(func.apply());
    } catch (RuntimeException e) {
      funcResult.completeExceptionally(e);
    }
    return funcResult
        .handle(
            (r, e) -> {
              if (e == null) {
                return WorkflowInternal.newPromise(r);
              }
              long elapsed = WorkflowInternal.currentTimeMillis() - startTime;
              long sleepTime = retryOptions.calculateSleepTime(attempt);
              if (retryOptions.shouldRethrow(e, attempt, elapsed, sleepTime)) {
                throw e;
              }
              // newTimer runs in a separate thread, so it performs trampolining eliminating tail
              // recursion.
              return WorkflowInternal.newTimer(Duration.ofMillis(sleepTime))
                  .thenCompose(
                      (nil) -> retryAsync(retryId, retryOptions, func, startTime, attempt + 1));
            })
        .thenCompose((r) -> r);
  }

  static <R> Promise<R> retryAsync(
      Functions.Func2<Integer, Long, Promise<R>> func, int attempt, long startTime) {

    CompletablePromise<R> funcResult = WorkflowInternal.newCompletablePromise();
    try {
      funcResult.completeFrom(func.apply(attempt, startTime));
    } catch (RuntimeException e) {
      funcResult.completeExceptionally(e);
    }

    return funcResult
        .handle(
            (r, e) -> {
              if (e == null) {
                return WorkflowInternal.newPromise(r);
              }

              if (!(e instanceof ActivityFailureException)) {
                throw e;
              }

              ActivityFailureException afe = (ActivityFailureException) e;

              if (afe.getBackoff() == null) {
                throw e;
              }

              // newTimer runs in a separate thread, so it performs trampolining eliminating tail
              // recursion.
              long nextStart = WorkflowInternal.currentTimeMillis() + afe.getBackoff().toMillis();
              return WorkflowInternal.newTimer(afe.getBackoff())
                  .thenCompose((nil) -> retryAsync(func, afe.getAttempt() + 1, nextStart));
            })
        .thenCompose((r) -> r);
  }

  private WorkflowRetryerInternal() {}
}

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.internal.sync;
19
20	import com.uber.cadence.common.RetryOptions;
21	import com.uber.cadence.workflow.ActivityFailureException;
22	import com.uber.cadence.workflow.CompletablePromise;
23	import com.uber.cadence.workflow.Functions;
24	import com.uber.cadence.workflow.Promise;
25	import com.uber.cadence.workflow.Workflow;
26	import java.time.Duration;
27
28	/**
29	* Implements operation retry logic for both synchronous and asynchronous operations. Internal
30	* class. Do not reference this class directly. Use {@link Workflow#retry(RetryOptions,
31	* Functions.Func)} or Async{@link #retry(RetryOptions, Functions.Func)}.
32	*/
33	final class WorkflowRetryerInternal {
34
35	public static <R> R validateOptionsAndRetry(RetryOptions options, Functions.Func<R> func) {
36	return retry(RetryOptions.merge(null, options), func);	×
37	}
38
39	/**
40	* Retry function synchronously.
41	*
42	* @param options retry options.
43	* @param func procedure to retry.
44	* @return result of func if ever completed successfully.
45	*/
46	public static <R> R retry(RetryOptions options, Functions.Func<R> func) {
47	options.validate();	1✔
48	int attempt = 1;	1✔
49	long startTime = WorkflowInternal.currentTimeMillis();	1✔
50	// Records retry options in the history allowing changing them without breaking determinism.
51	String retryId = WorkflowInternal.randomUUID().toString();	1✔
52	RetryOptions retryOptions =	1✔
53	WorkflowInternal.mutableSideEffect(	1✔
54	retryId, RetryOptions.class, RetryOptions.class, Object::equals, () -> options);	1✔
55	while (true) {
56	long nextSleepTime = retryOptions.calculateSleepTime(attempt);	1✔
57	try {
58	return func.apply();	×
59	} catch (Exception e) {	1✔
60	long elapsed = WorkflowInternal.currentTimeMillis() - startTime;	1✔
61	if (retryOptions.shouldRethrow(e, attempt, elapsed, nextSleepTime)) {	1✔
62	throw WorkflowInternal.wrap(e);	1✔
63	}
64	}
65	attempt++;	1✔
66	WorkflowInternal.sleep(Duration.ofMillis(nextSleepTime));	1✔
67	}	1✔
68	}
69
70	/**
71	* Retry function asynchronously.
72	*
73	* @param options retry options.
74	* @param func procedure to retry.
75	* @return result promise to the result or failure if retries stopped according to options.
76	*/
77	public static <R> Promise<R> retryAsync(RetryOptions options, Functions.Func<Promise<R>> func) {
78	String retryId = WorkflowInternal.randomUUID().toString();	1✔
79	long startTime = WorkflowInternal.currentTimeMillis();	1✔
80	return retryAsync(retryId, options, func, startTime, 1);	1✔
81	}
82
83	private static <R> Promise<R> retryAsync(
84	String retryId,
85	RetryOptions options,
86	Functions.Func<Promise<R>> func,
87	long startTime,
88	long attempt) {
89	options.validate();	1✔
90	RetryOptions retryOptions =	1✔
91	WorkflowInternal.mutableSideEffect(	1✔
92	retryId, RetryOptions.class, RetryOptions.class, Object::equals, () -> options);	1✔
93
94	CompletablePromise<R> funcResult = WorkflowInternal.newCompletablePromise();	1✔
95	try {
96	funcResult.completeFrom(func.apply());	1✔
97	} catch (RuntimeException e) {	×
98	funcResult.completeExceptionally(e);	×
99	}	1✔
100	return funcResult	1✔
101	.handle(	1✔
102	(r, e) -> {
103	if (e == null) {	1✔
104	return WorkflowInternal.newPromise(r);	×
105	}
106	long elapsed = WorkflowInternal.currentTimeMillis() - startTime;	1✔
107	long sleepTime = retryOptions.calculateSleepTime(attempt);	1✔
108	if (retryOptions.shouldRethrow(e, attempt, elapsed, sleepTime)) {	1✔
109	throw e;	1✔
110	}
111	// newTimer runs in a separate thread, so it performs trampolining eliminating tail
112	// recursion.
113	return WorkflowInternal.newTimer(Duration.ofMillis(sleepTime))	1✔
114	.thenCompose(	1✔
115	(nil) -> retryAsync(retryId, retryOptions, func, startTime, attempt + 1));	1✔
116	})
117	.thenCompose((r) -> r);	1✔
118	}
119
120	static <R> Promise<R> retryAsync(
121	Functions.Func2<Integer, Long, Promise<R>> func, int attempt, long startTime) {
122
123	CompletablePromise<R> funcResult = WorkflowInternal.newCompletablePromise();	1✔
124	try {
125	funcResult.completeFrom(func.apply(attempt, startTime));	1✔
126	} catch (RuntimeException e) {	×
127	funcResult.completeExceptionally(e);	×
128	}	1✔
129
130	return funcResult	1✔
131	.handle(	1✔
132	(r, e) -> {
133	if (e == null) {	1✔
134	return WorkflowInternal.newPromise(r);	1✔
135	}
136
137	if (!(e instanceof ActivityFailureException)) {	1✔
138	throw e;	×
139	}
140
141	ActivityFailureException afe = (ActivityFailureException) e;	1✔
142
143	if (afe.getBackoff() == null) {	1✔
144	throw e;	1✔
145	}
146
147	// newTimer runs in a separate thread, so it performs trampolining eliminating tail
148	// recursion.
149	long nextStart = WorkflowInternal.currentTimeMillis() + afe.getBackoff().toMillis();	1✔
150	return WorkflowInternal.newTimer(afe.getBackoff())	1✔
151	.thenCompose((nil) -> retryAsync(func, afe.getAttempt() + 1, nextStart));	1✔
152	})
153	.thenCompose((r) -> r);	1✔
154	}
155
156	private WorkflowRetryerInternal() {}
157	}

uber / cadence-java-client / 2555

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