Cleanup retryer and retry options (#558)

Author: meiliang86

src/main/java/com/uber/cadence/client/WorkflowClientOptions.java

@@ -1,8 +1,8 @@
 /*
+ *  Modifications Copyright (c) 2017-2020 Uber Technologies Inc.
+ *  Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
  *  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

src/main/java/com/uber/cadence/internal/common/Retryer.java renamed to src/main/java/com/uber/cadence/internal/common/RpcRetryer.java

@@ -1,8 +1,8 @@
 /*
- *  Modifications Copyright (c) 2017-2020 Uber Technologies Inc.
- *  Portions of the Software are attributed to Copyright (c) 2020 Temporal Technologies Inc.
  *  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
@@ -19,7 +19,13 @@
 
 import static com.uber.cadence.internal.common.CheckedExceptionWrapper.unwrap;
 
-import com.uber.cadence.*;
+import com.uber.cadence.BadRequestError;
+import com.uber.cadence.CancellationAlreadyRequestedError;
+import com.uber.cadence.DomainAlreadyExistsError;
+import com.uber.cadence.DomainNotActiveError;
+import com.uber.cadence.EntityNotExistsError;
+import com.uber.cadence.QueryFailedError;
+import com.uber.cadence.WorkflowExecutionAlreadyStartedError;
 import com.uber.cadence.common.RetryOptions;
 import java.time.Duration;
 import java.util.concurrent.CancellationException;
@@ -31,8 +37,8 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public final class Retryer {
-  public static final RetryOptions DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS;
+public final class RpcRetryer {
+  public static final RetryOptions DEFAULT_RPC_RETRY_OPTIONS;
 
   private static final Duration RETRY_SERVICE_OPERATION_INITIAL_INTERVAL = Duration.ofMillis(20);
   private static final Duration RETRY_SERVICE_OPERATION_EXPIRATION_INTERVAL = Duration.ofMinutes(1);
@@ -58,7 +64,7 @@ public final class Retryer {
         QueryFailedError.class,
         DomainNotActiveError.class,
         CancellationAlreadyRequestedError.class);
-    DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS = roBuilder.validateBuildWithDefaults();
+    DEFAULT_RPC_RETRY_OPTIONS = roBuilder.validateBuildWithDefaults();
   }
 
   public interface RetryableProc<E extends Throwable> {
@@ -81,7 +87,7 @@ private static class ValueExceptionPair<V> {
     private final CompletableFuture<V> value;
     private final Throwable exception;
 
-    public ValueExceptionPair(CompletableFuture<V> value, Throwable exception) {
+    ValueExceptionPair(CompletableFuture<V> value, Throwable exception) {
       this.value = value;
       this.exception = exception;
     }
@@ -95,7 +101,7 @@ public Throwable getException() {
     }
   }
 
-  private static final Logger log = LoggerFactory.getLogger(Retryer.class);
+  private static final Logger log = LoggerFactory.getLogger(RpcRetryer.class);
 
   public static <T extends Throwable> void retry(RetryOptions options, RetryableProc<T> r)
       throws T {
@@ -107,6 +113,10 @@ public static <T extends Throwable> void retry(RetryOptions options, RetryablePr
         });
   }
 
+  public static <T extends Throwable> void retry(RetryableProc<T> r) throws T {
+    retry(DEFAULT_RPC_RETRY_OPTIONS, r);
+  }
+
   public static <R, T extends Throwable> R retryWithResult(
       RetryOptions options, RetryableFunc<R, T> r) throws T {
     int attempt = 0;
@@ -266,5 +276,5 @@ private static <T extends Throwable> void rethrow(Exception e) throws T {
   }
 
   /** Prohibits instantiation. */
-  private Retryer() {}
+  private RpcRetryer() {}
 }

src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java

@@ -27,7 +27,6 @@
 import com.google.gson.JsonParser;
 import com.google.gson.JsonPrimitive;
 import com.uber.cadence.ActivityType;
-import com.uber.cadence.BadRequestError;
 import com.uber.cadence.Decision;
 import com.uber.cadence.DecisionType;
 import com.uber.cadence.DescribeWorkflowExecutionRequest;
@@ -91,15 +90,6 @@ public class WorkflowExecutionUtils {
    */
   private static final String INDENTATION = "  ";
 
-  private static RetryOptions retryParameters =
-      new RetryOptions.Builder()
-          .setBackoffCoefficient(2)
-          .setInitialInterval(Duration.ofMillis(500))
-          .setMaximumInterval(Duration.ofSeconds(30))
-          .setMaximumAttempts(Integer.MAX_VALUE)
-          .setDoNotRetry(BadRequestError.class, EntityNotExistsError.class)
-          .build();
-
   // Wait period for passive cluster to retry getting workflow result in case of replication delay.
   private static final long ENTITY_NOT_EXIST_RETRY_WAIT_MILLIS = 500;
 
@@ -216,7 +206,7 @@ private static HistoryEvent getInstanceCloseEvent(
       RetryOptions retryOptions = getRetryOptionWithTimeout(timeout, unit);
       try {
         response =
-            Retryer.retryWithResult(
+            RpcRetryer.retryWithResult(
                 retryOptions,
                 () -> service.GetWorkflowExecutionHistoryWithTimeout(r, unit.toMillis(timeout)));
       } catch (EntityNotExistsError e) {
@@ -343,7 +333,7 @@ private static CompletableFuture<HistoryEvent> getInstanceCloseEventAsync(
   }
 
   private static RetryOptions getRetryOptionWithTimeout(long timeout, TimeUnit unit) {
-    return new RetryOptions.Builder(retryParameters)
+    return new RetryOptions.Builder(RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS)
         .setExpiration(Duration.ofSeconds(unit.toSeconds(timeout)))
         .build();
   }
@@ -355,7 +345,7 @@ private static RetryOptions getRetryOptionWithTimeout(long timeout, TimeUnit uni
           long timeout,
           TimeUnit unit) {
     RetryOptions retryOptions = getRetryOptionWithTimeout(timeout, unit);
-    return Retryer.retryWithResultAsync(
+    return RpcRetryer.retryWithResultAsync(
         retryOptions,
         () -> {
           CompletableFuture<GetWorkflowExecutionHistoryResponse> result = new CompletableFuture<>();

src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java

@@ -118,9 +118,8 @@ private WorkflowExecution startWorkflowInternal(StartWorkflowExecutionParameters
     StartWorkflowExecutionResponse result;
     try {
       result =
-          Retryer.retryWithResult(
-              Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
-              () -> service.StartWorkflowExecution(request));
+          RpcRetryer.retryWithResult(
+              RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS, () -> service.StartWorkflowExecution(request));
     } catch (WorkflowExecutionAlreadyStartedError e) {
       throw e;
     } catch (TException e) {
@@ -137,7 +136,7 @@ private RetryOptions getRetryOptionsWithExpiration(RetryOptions o, Long timeoutI
     if (timeoutInMillis == null || timeoutInMillis <= 0 || timeoutInMillis == Long.MAX_VALUE) {
       return o;
     }
-    return new RetryOptions.Builder(Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS)
+    return new RetryOptions.Builder(RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS)
         .setExpiration(Duration.ofMillis((timeoutInMillis)))
         .build();
   }
@@ -146,9 +145,8 @@ private CompletableFuture<WorkflowExecution> startWorkflowAsyncInternal(
       StartWorkflowExecutionParameters startParameters, Long timeoutInMillis) {
     StartWorkflowExecutionRequest request = getStartRequest(startParameters);
 
-    return Retryer.retryWithResultAsync(
-        getRetryOptionsWithExpiration(
-            Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS, timeoutInMillis),
+    return RpcRetryer.retryWithResultAsync(
+        getRetryOptionsWithExpiration(RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS, timeoutInMillis),
         () -> {
           CompletableFuture<WorkflowExecution> result = new CompletableFuture<>();
           try {
@@ -272,9 +270,7 @@ public void signalWorkflowExecution(SignalExternalWorkflowParameters signalParam
     SignalWorkflowExecutionRequest request = getSignalRequest(signalParameters);
 
     try {
-      Retryer.retry(
-          Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
-          () -> service.SignalWorkflowExecution(request));
+      RpcRetryer.retry(() -> service.SignalWorkflowExecution(request));
     } catch (TException e) {
       throw CheckedExceptionWrapper.wrap(e);
     }
@@ -290,9 +286,8 @@ public CompletableFuture<Void> signalWorkflowExecutionAsync(
   public CompletableFuture<Void> signalWorkflowExecutionAsync(
       SignalExternalWorkflowParameters signalParameters, Long timeoutInMillis) {
     SignalWorkflowExecutionRequest request = getSignalRequest(signalParameters);
-    return Retryer.retryWithResultAsync(
-        getRetryOptionsWithExpiration(
-            Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS, timeoutInMillis),
+    return RpcRetryer.retryWithResultAsync(
+        getRetryOptionsWithExpiration(RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS, timeoutInMillis),
         () -> {
           CompletableFuture<Void> result = new CompletableFuture<>();
           try {
@@ -387,8 +382,8 @@ private WorkflowExecution signalWithStartWorkflowInternal(
     StartWorkflowExecutionResponse result;
     try {
       result =
-          Retryer.retryWithResult(
-              Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
+          RpcRetryer.retryWithResult(
+              RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS,
               () -> service.SignalWithStartWorkflowExecution(request));
     } catch (TException e) {
       throw CheckedExceptionWrapper.wrap(e);
@@ -405,9 +400,7 @@ public void requestCancelWorkflowExecution(WorkflowExecution execution) {
     request.setDomain(domain);
     request.setWorkflowExecution(execution);
     try {
-      Retryer.retry(
-          Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
-          () -> service.RequestCancelWorkflowExecution(request));
+      RpcRetryer.retry(() -> service.RequestCancelWorkflowExecution(request));
     } catch (TException e) {
       throw CheckedExceptionWrapper.wrap(e);
     }
@@ -427,9 +420,8 @@ public QueryWorkflowResponse queryWorkflow(QueryWorkflowParameters queryParamete
     request.setQueryRejectCondition(queryParameters.getQueryRejectCondition());
     try {
       QueryWorkflowResponse response =
-          Retryer.retryWithResult(
-              Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
-              () -> service.QueryWorkflow(request));
+          RpcRetryer.retryWithResult(
+              RpcRetryer.DEFAULT_RPC_RETRY_OPTIONS, () -> service.QueryWorkflow(request));
       return response;
     } catch (TException e) {
       throw CheckedExceptionWrapper.wrap(e);
@@ -451,9 +443,7 @@ public void terminateWorkflowExecution(TerminateWorkflowExecutionParameters term
     request.setReason(terminateParameters.getReason());
     //        request.setChildPolicy(terminateParameters.getChildPolicy());
     try {
-      Retryer.retry(
-          Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
-          () -> service.TerminateWorkflowExecution(request));
+      RpcRetryer.retry(() -> service.TerminateWorkflowExecution(request));
     } catch (TException e) {
       throw CheckedExceptionWrapper.wrap(e);
     }

src/main/java/com/uber/cadence/internal/external/ManualActivityCompletionClientImpl.java

@@ -31,7 +31,7 @@
 import com.uber.cadence.client.ActivityCompletionFailureException;
 import com.uber.cadence.client.ActivityNotExistsException;
 import com.uber.cadence.converter.DataConverter;
-import com.uber.cadence.internal.common.Retryer;
+import com.uber.cadence.internal.common.RpcRetryer;
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.serviceclient.IWorkflowService;
 import com.uber.m3.tally.Scope;
@@ -95,9 +95,7 @@ public void complete(Object result) {
       request.setResult(convertedResult);
       request.setTaskToken(taskToken);
       try {
-        Retryer.retry(
-            Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
-            () -> service.RespondActivityTaskCompleted(request));
+        RpcRetryer.retry(() -> service.RespondActivityTaskCompleted(request));
         metricsScope.counter(MetricsType.ACTIVITY_TASK_COMPLETED_COUNTER).inc(1);
       } catch (EntityNotExistsError e) {
         throw new ActivityNotExistsException(e);
@@ -139,9 +137,7 @@ public void fail(Throwable failure) {
       request.setDetails(dataConverter.toData(failure));
       request.setTaskToken(taskToken);
       try {
-        Retryer.retry(
-            Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
-            () -> service.RespondActivityTaskFailed(request));
+        RpcRetryer.retry(() -> service.RespondActivityTaskFailed(request));
         metricsScope.counter(MetricsType.ACTIVITY_TASK_FAILED_COUNTER).inc(1);
       } catch (EntityNotExistsError e) {
         throw new ActivityNotExistsException(e);
@@ -156,9 +152,7 @@ public void fail(Throwable failure) {
       request.setWorkflowID(execution.getWorkflowId());
       request.setRunID(execution.getRunId());
       try {
-        Retryer.retry(
-            Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS,
-            () -> service.RespondActivityTaskFailedByID(request));
+        RpcRetryer.retry(() -> service.RespondActivityTaskFailedByID(request));
         metricsScope.counter(MetricsType.ACTIVITY_TASK_FAILED_BY_ID_COUNTER).inc(1);
       } catch (EntityNotExistsError e) {
         throw new ActivityNotExistsException(e);

src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java

@@ -32,7 +32,7 @@
 import com.uber.cadence.WorkflowType;
 import com.uber.cadence.common.RetryOptions;
 import com.uber.cadence.internal.common.OptionsUtils;
-import com.uber.cadence.internal.common.Retryer;
+import com.uber.cadence.internal.common.RpcRetryer;
 import com.uber.cadence.internal.metrics.MetricsTag;
 import com.uber.cadence.internal.metrics.MetricsType;
 import com.uber.cadence.internal.replay.HistoryHelper.DecisionEvents;
@@ -667,7 +667,7 @@ public HistoryEvent next() {
 
           try {
             GetWorkflowExecutionHistoryResponse r =
-                Retryer.retryWithResult(
+                RpcRetryer.retryWithResult(
                     retryOptions, () -> service.GetWorkflowExecutionHistory(request));
             current = r.getHistory().getEventsIterator();
             nextPageToken = r.getNextPageToken();

src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java

@@ -113,7 +113,7 @@ public DecisionTaskHandler.Result handleDecisionTask(PollForDecisionTaskResponse
       e.printStackTrace(pw);
       String stackTrace = sw.toString();
       failedRequest.setDetails(stackTrace.getBytes(StandardCharsets.UTF_8));
-      return new DecisionTaskHandler.Result(null, failedRequest, null, null);
+      return new DecisionTaskHandler.Result(null, failedRequest, null);
     }
   }
 
@@ -236,7 +236,7 @@ private Result processQuery(PollForDecisionTaskResponse decisionTask) {
         cache.markProcessingDone(decisionTask);
       }
     }
-    return new Result(null, null, queryCompletedRequest, null);
+    return new Result(null, null, queryCompletedRequest);
   }
 
   private Result createCompletedRequest(
@@ -254,7 +254,7 @@ private Result createCompletedRequest(
           (int) stickyTaskListScheduleToStartTimeout.getSeconds());
       completedRequest.setStickyAttributes(attributes);
     }
-    return new Result(completedRequest, null, null, null);
+    return new Result(completedRequest, null, null);
   }
 
   @Override

src/main/java/com/uber/cadence/internal/sync/POJOActivityTaskHandler.java

@@ -147,8 +147,7 @@ private ActivityTaskHandler.Result mapToActivityFailure(
     failure = CheckedExceptionWrapper.unwrap(failure);
     result.setReason(failure.getClass().getName());
     result.setDetails(dataConverter.toData(failure));
-    return new ActivityTaskHandler.Result(
-        null, new Result.TaskFailedResult(result, failure), null, null);
+    return new ActivityTaskHandler.Result(null, new Result.TaskFailedResult(result, failure), null);
   }
 
   @Override
@@ -214,12 +213,12 @@ public ActivityTaskHandler.Result execute(ActivityTaskImpl task, Scope metricsSc
         Object result = method.invoke(activity, args);
         RespondActivityTaskCompletedRequest request = new RespondActivityTaskCompletedRequest();
         if (context.isDoNotCompleteOnReturn()) {
-          return new ActivityTaskHandler.Result(null, null, null, null);
+          return new ActivityTaskHandler.Result(null, null, null);
         }
         if (method.getReturnType() != Void.TYPE) {
           request.setResult(dataConverter.toData(result));
         }
-        return new ActivityTaskHandler.Result(request, null, null, null);
+        return new ActivityTaskHandler.Result(request, null, null);
       } catch (RuntimeException | IllegalAccessException e) {
         return mapToActivityFailure(e, metricsScope, false);
       } catch (InvocationTargetException e) {
@@ -252,7 +251,7 @@ public ActivityTaskHandler.Result execute(ActivityTaskImpl task, Scope metricsSc
         if (method.getReturnType() != Void.TYPE) {
           request.setResult(dataConverter.toData(result));
         }
-        return new ActivityTaskHandler.Result(request, null, null, null);
+        return new ActivityTaskHandler.Result(request, null, null);
       } catch (RuntimeException | IllegalAccessException e) {
         return mapToActivityFailure(e, metricsScope, true);
       } catch (InvocationTargetException e) {

src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java

@@ -139,9 +139,6 @@ public Worker newWorker(
     WorkerOptions.Builder builder =
         WorkerOptions.newBuilder()
             .setInterceptorFactory(testEnvironmentOptions.getInterceptorFactory());
-    if (testEnvironmentOptions.getDataConverter() != null) {
-      builder.setDataConverter(testEnvironmentOptions.getDataConverter());
-    }
     builder = overrideOptions.apply(builder);
     return workerFactory.newWorker(taskList, builder.build());
   }