diff --git a/src/main/java/com/uber/cadence/activity/ActivityInterface.java b/src/main/java/com/uber/cadence/activity/ActivityInterface.java
new file mode 100644
index 000000000..81b6cc6f8
--- /dev/null
+++ b/src/main/java/com/uber/cadence/activity/ActivityInterface.java
@@ -0,0 +1,76 @@
+/*
+ * 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.
+ *
+ * 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.activity;
+
+import com.uber.cadence.workflow.Workflow;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Indicates that the interface is an activity interface. Only interfaces annotated with this
+ * annotation can be used as parameters to {@link Workflow#newActivityStub(Class)} methods.
+ *
+ * Each method of the interface annotated with ActivityInterface including inherited
+ * from interfaces is a separate activity. By default the name of an activity type is "short
+ * interface name"_"method name".
+ *
+ *
Example:
+ *
+ *
+ * public interface A {
+ * a();
+ * }
+ *
+ * {@literal @}ActivityInterface
+ * public interface B extends A {
+ * b();
+ * }
+ *
+ * {@literal @}ActivityInterface
+ * public interface C extends B {
+ * c();
+ * }
+ *
+ * public class CImpl implements C {
+ * public void a() {}
+ * public void b() {}
+ * public void c() {}
+ * }
+ *
+ *
+ * When CImpl instance is registered with the {@link com.uber.cadence.worker.Worker}
+ * the following activities are registered:
+ *
+ *
+ *
+ *
+ * - B_a
+ *
- B_b
+ *
- C_c
+ *
+ *
+ * Note that method a() is registered as "B_a" because interface A lacks
+ * ActivityInterface annotation. The workflow code can call activities through stubs to B
+ * and C interfaces. A call to crate stub to A interface will fail
+ * as A is not annotated with ActivityInterface.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+public @interface ActivityInterface {}
diff --git a/src/main/java/com/uber/cadence/activity/ActivityMethod.java b/src/main/java/com/uber/cadence/activity/ActivityMethod.java
index 239492a45..00118830a 100644
--- a/src/main/java/com/uber/cadence/activity/ActivityMethod.java
+++ b/src/main/java/com/uber/cadence/activity/ActivityMethod.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
@@ -21,12 +21,11 @@
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
+import java.time.Duration;
/**
* Indicates that the method is an activity method. This annotation applies only to activity
- * interface methods. Not required. Use it to override default activity type name or other options.
- * When both {@link ActivityOptions} and {@link ActivityMethod} have non default value for some
- * parameter the {@link ActivityOptions} one takes precedence.
+ * interface methods. Not required. Use it to override default activity type name.
*/
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
@@ -40,30 +39,40 @@
* task list (use {@link #scheduleToStartTimeoutSeconds()} to limit it) plus activity execution
* time (use {@link #startToCloseTimeoutSeconds()} to limit it). Either this option or both
* schedule to start and start to close are required.
+ *
+ * @deprecated use {@link ActivityOptions.Builder#setScheduleToCloseTimeout(Duration)} instead.
*/
int scheduleToCloseTimeoutSeconds() default 0;
/**
* Time activity can stay in task list before it is picked up by a worker. If schedule to close is
* not provided then both this and start to close are required.
+ *
+ * @deprecated use {@link ActivityOptions.Builder#setScheduleToStartTimeout(Duration)} instead.
*/
int scheduleToStartTimeoutSeconds() default 0;
/**
* Maximum activity execution time after it was sent to a worker. If schedule to close is not
* provided then both this and schedule to start are required.
+ *
+ * @deprecated use {@link ActivityOptions.Builder#setStartToCloseTimeout(Duration)} instead.
*/
int startToCloseTimeoutSeconds() default 0;
/**
* Heartbeat interval. Activity must heartbeat before this interval passes after a last heartbeat
* or activity start.
+ *
+ * @deprecated use {@link ActivityOptions.Builder#setHeartbeatTimeout(Duration)} instead.
*/
int heartbeatTimeoutSeconds() default 0;
/**
* Task list to use when dispatching activity task to a worker. By default it is the same task
* list name the workflow was started with.
+ *
+ * @deprecated use {@link ActivityOptions.Builder#setTaskList(String)} instead.
*/
String taskList() default "";
}
diff --git a/src/main/java/com/uber/cadence/activity/ActivityOptions.java b/src/main/java/com/uber/cadence/activity/ActivityOptions.java
index da3062c18..d0b6016a4 100644
--- a/src/main/java/com/uber/cadence/activity/ActivityOptions.java
+++ b/src/main/java/com/uber/cadence/activity/ActivityOptions.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
@@ -29,36 +29,22 @@
/** Options used to configure how an activity is invoked. */
public final class ActivityOptions {
- /**
- * Used to merge annotation and options. Options takes precedence. Returns options with all
- * defaults filled in.
- */
- public static ActivityOptions merge(ActivityMethod a, MethodRetry r, ActivityOptions o) {
- if (a == null) {
- if (r == null) {
- return new ActivityOptions.Builder(o).validateAndBuildWithDefaults();
- }
- RetryOptions mergedR = RetryOptions.merge(r, o.getRetryOptions());
- return new ActivityOptions.Builder().setRetryOptions(mergedR).validateAndBuildWithDefaults();
- }
- if (o == null) {
- o = new ActivityOptions.Builder().build();
- }
- return new ActivityOptions.Builder()
- .setScheduleToCloseTimeout(
- mergeDuration(a.scheduleToCloseTimeoutSeconds(), o.getScheduleToCloseTimeout()))
- .setScheduleToStartTimeout(
- mergeDuration(a.scheduleToStartTimeoutSeconds(), o.getScheduleToStartTimeout()))
- .setStartToCloseTimeout(
- mergeDuration(a.startToCloseTimeoutSeconds(), o.getStartToCloseTimeout()))
- .setHeartbeatTimeout(mergeDuration(a.heartbeatTimeoutSeconds(), o.getHeartbeatTimeout()))
- .setTaskList(
- o.getTaskList() != null
- ? o.getTaskList()
- : (a.taskList().isEmpty() ? null : a.taskList()))
- .setRetryOptions(RetryOptions.merge(r, o.getRetryOptions()))
- .setContextPropagators(o.getContextPropagators())
- .validateAndBuildWithDefaults();
+ public static Builder newBuilder() {
+ return new Builder();
+ }
+
+ public static Builder newBuilder(ActivityOptions options) {
+ return new Builder(options);
+ }
+
+ public static ActivityOptions getDefaultInstance() {
+ return DEFAULT_INSTANCE;
+ }
+
+ private static final ActivityOptions DEFAULT_INSTANCE;
+
+ static {
+ DEFAULT_INSTANCE = ActivityOptions.newBuilder().build();
}
public static final class Builder {
@@ -155,6 +141,14 @@ public Builder setContextPropagators(List contextPropagators)
return this;
}
+ /**
+ * Properties that are set on this builder take precedence over ones found in the annotation.
+ */
+ public Builder setMethodRetry(MethodRetry r) {
+ retryOptions = RetryOptions.merge(r, retryOptions);
+ return this;
+ }
+
public ActivityOptions build() {
return new ActivityOptions(
heartbeatTimeout,
@@ -324,15 +318,4 @@ public int hashCode() {
retryOptions,
contextPropagators);
}
-
- static Duration mergeDuration(int annotationSeconds, Duration options) {
- if (options == null) {
- if (annotationSeconds == 0) {
- return null;
- }
- return Duration.ofSeconds(annotationSeconds);
- } else {
- return options;
- }
- }
}
diff --git a/src/main/java/com/uber/cadence/activity/LocalActivityOptions.java b/src/main/java/com/uber/cadence/activity/LocalActivityOptions.java
index 9534af9a2..dde8334b8 100644
--- a/src/main/java/com/uber/cadence/activity/LocalActivityOptions.java
+++ b/src/main/java/com/uber/cadence/activity/LocalActivityOptions.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
@@ -29,31 +29,23 @@
/** Options used to configure how an local activity is invoked. */
public final class LocalActivityOptions {
- /**
- * Used to merge annotation and options. Options takes precedence. Returns options with all
- * defaults filled in.
- */
- public static LocalActivityOptions merge(
- ActivityMethod a, MethodRetry r, LocalActivityOptions o) {
- if (a == null) {
- if (r == null) {
- return new LocalActivityOptions.Builder(o).validateAndBuildWithDefaults();
- }
- RetryOptions mergedR = RetryOptions.merge(r, o.getRetryOptions());
- return new LocalActivityOptions.Builder()
- .setRetryOptions(mergedR)
- .validateAndBuildWithDefaults();
- }
- if (o == null) {
- o = new LocalActivityOptions.Builder().build();
- }
- return new LocalActivityOptions.Builder()
- .setScheduleToCloseTimeout(
- ActivityOptions.mergeDuration(
- a.scheduleToCloseTimeoutSeconds(), o.getScheduleToCloseTimeout()))
- .setRetryOptions(RetryOptions.merge(r, o.getRetryOptions()))
- .setContextPropagators(o.getContextPropagators())
- .validateAndBuildWithDefaults();
+ public static Builder newBuilder() {
+ return new Builder(null);
+ }
+
+ /** @param o null is allowed */
+ public static Builder newBuilder(LocalActivityOptions o) {
+ return new Builder(o);
+ }
+
+ public static LocalActivityOptions getDefaultInstance() {
+ return DEFAULT_INSTANCE;
+ }
+
+ private static final LocalActivityOptions DEFAULT_INSTANCE;
+
+ static {
+ DEFAULT_INSTANCE = LocalActivityOptions.newBuilder().build();
}
public static final class Builder {
@@ -92,6 +84,17 @@ public Builder setContextPropagators(List contextPropagators)
return this;
}
+ /**
+ * Merges MethodRetry annotation. The values of this builder take precedence over annotation
+ * ones.
+ */
+ public Builder setMethodRetry(MethodRetry r) {
+ if (r != null) {
+ this.retryOptions = RetryOptions.merge(r, retryOptions);
+ }
+ return this;
+ }
+
public LocalActivityOptions build() {
return new LocalActivityOptions(scheduleToCloseTimeout, retryOptions, contextPropagators);
}
diff --git a/src/main/java/com/uber/cadence/client/WorkflowClientOptions.java b/src/main/java/com/uber/cadence/client/WorkflowClientOptions.java
index d47ef8528..df59f25eb 100644
--- a/src/main/java/com/uber/cadence/client/WorkflowClientOptions.java
+++ b/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
diff --git a/src/main/java/com/uber/cadence/internal/common/InternalUtils.java b/src/main/java/com/uber/cadence/internal/common/InternalUtils.java
index 73498885a..5b1d0a1c5 100644
--- a/src/main/java/com/uber/cadence/internal/common/InternalUtils.java
+++ b/src/main/java/com/uber/cadence/internal/common/InternalUtils.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
@@ -51,7 +51,11 @@ public final class InternalUtils {
* @return "Simple class name"::"methodName"
*/
public static String getSimpleName(Method method) {
- return method.getDeclaringClass().getSimpleName() + "::" + method.getName();
+ return getSimpleName(method.getDeclaringClass(), method);
+ }
+
+ public static String getSimpleName(Class type, Method method) {
+ return type.getSimpleName() + "::" + method.getName();
}
public static String getWorkflowType(Method method, WorkflowMethod workflowMethod) {
diff --git a/src/main/java/com/uber/cadence/internal/common/Retryer.java b/src/main/java/com/uber/cadence/internal/common/RpcRetryer.java
similarity index 91%
rename from src/main/java/com/uber/cadence/internal/common/Retryer.java
rename to src/main/java/com/uber/cadence/internal/common/RpcRetryer.java
index a2d7e0a9a..4f27f6f87 100644
--- a/src/main/java/com/uber/cadence/internal/common/Retryer.java
+++ b/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 {
@@ -81,7 +87,7 @@ private static class ValueExceptionPair {
private final CompletableFuture value;
private final Throwable exception;
- public ValueExceptionPair(CompletableFuture value, Throwable exception) {
+ ValueExceptionPair(CompletableFuture 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 void retry(RetryOptions options, RetryableProc r)
throws T {
@@ -107,6 +113,10 @@ public static void retry(RetryOptions options, RetryablePr
});
}
+ public static void retry(RetryableProc r) throws T {
+ retry(DEFAULT_RPC_RETRY_OPTIONS, r);
+ }
+
public static R retryWithResult(
RetryOptions options, RetryableFunc r) throws T {
int attempt = 0;
@@ -266,5 +276,5 @@ private static void rethrow(Exception e) throws T {
}
/** Prohibits instantiation. */
- private Retryer() {}
+ private RpcRetryer() {}
}
diff --git a/src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java b/src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java
index 02bf2cc24..5f37a16df 100644
--- a/src/main/java/com/uber/cadence/internal/common/WorkflowExecutionUtils.java
+++ b/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 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 result = new CompletableFuture<>();
diff --git a/src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java b/src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java
index 6d86d186a..ebcfc67da 100644
--- a/src/main/java/com/uber/cadence/internal/external/GenericWorkflowClientExternalImpl.java
+++ b/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 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 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 signalWorkflowExecutionAsync(
public CompletableFuture 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 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);
}
diff --git a/src/main/java/com/uber/cadence/internal/external/ManualActivityCompletionClientImpl.java b/src/main/java/com/uber/cadence/internal/external/ManualActivityCompletionClientImpl.java
index ab30159bc..3367e705a 100644
--- a/src/main/java/com/uber/cadence/internal/external/ManualActivityCompletionClientImpl.java
+++ b/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);
diff --git a/src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java b/src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java
index d019005e0..b3ddcf760 100644
--- a/src/main/java/com/uber/cadence/internal/replay/ReplayDecider.java
+++ b/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();
diff --git a/src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java b/src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java
index d06a0c01b..c1c6d9b05 100644
--- a/src/main/java/com/uber/cadence/internal/replay/ReplayDecisionTaskHandler.java
+++ b/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
diff --git a/src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandler.java b/src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandler.java
index 17836bb5f..2bf87603e 100644
--- a/src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandler.java
+++ b/src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandler.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
@@ -17,6 +17,7 @@
package com.uber.cadence.internal.sync;
+import com.google.common.base.Strings;
import com.uber.cadence.activity.ActivityMethod;
import com.uber.cadence.activity.ActivityOptions;
import com.uber.cadence.common.MethodRetry;
@@ -24,6 +25,7 @@
import com.uber.cadence.workflow.WorkflowInterceptor;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
+import java.time.Duration;
import java.util.function.Function;
class ActivityInvocationHandler extends ActivityInvocationHandlerBase {
@@ -31,20 +33,48 @@ class ActivityInvocationHandler extends ActivityInvocationHandlerBase {
private final WorkflowInterceptor activityExecutor;
static InvocationHandler newInstance(
- ActivityOptions options, WorkflowInterceptor activityExecutor) {
- return new ActivityInvocationHandler(options, activityExecutor);
+ Class activityInterface, ActivityOptions options, WorkflowInterceptor activityExecutor) {
+ return new ActivityInvocationHandler(activityInterface, activityExecutor, options);
}
- private ActivityInvocationHandler(ActivityOptions options, WorkflowInterceptor activityExecutor) {
+ private ActivityInvocationHandler(
+ Class activityInterface, WorkflowInterceptor activityExecutor, ActivityOptions options) {
this.options = options;
this.activityExecutor = activityExecutor;
+ init(activityInterface);
}
+ @SuppressWarnings("deprecation")
@Override
protected Function getActivityFunc(
Method method, MethodRetry methodRetry, ActivityMethod activityMethod, String activityName) {
Function function;
- ActivityOptions mergedOptions = ActivityOptions.merge(activityMethod, methodRetry, options);
+ ActivityOptions.Builder optionsBuilder =
+ ActivityOptions.newBuilder(options).setMethodRetry(methodRetry);
+ if (activityMethod != null) {
+ // options always take precedence over activity method annotation.
+ if (options.getStartToCloseTimeout() == null) {
+ optionsBuilder.setStartToCloseTimeout(
+ Duration.ofSeconds(activityMethod.startToCloseTimeoutSeconds()));
+ }
+ if (options.getScheduleToStartTimeout() == null) {
+ optionsBuilder.setScheduleToStartTimeout(
+ Duration.ofSeconds(activityMethod.scheduleToStartTimeoutSeconds()));
+ }
+ if (options.getScheduleToCloseTimeout() == null) {
+ optionsBuilder.setScheduleToCloseTimeout(
+ Duration.ofSeconds(activityMethod.scheduleToCloseTimeoutSeconds()));
+ }
+ if (options.getHeartbeatTimeout() == null) {
+ optionsBuilder.setHeartbeatTimeout(
+ Duration.ofSeconds(activityMethod.heartbeatTimeoutSeconds()));
+ }
+ if (Strings.isNullOrEmpty(options.getTaskList())
+ && !Strings.isNullOrEmpty(activityMethod.taskList())) {
+ optionsBuilder.setTaskList(activityMethod.taskList());
+ }
+ }
+ ActivityOptions mergedOptions = optionsBuilder.build();
ActivityStub stub = ActivityStubImpl.newInstance(mergedOptions, activityExecutor);
function =
diff --git a/src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandlerBase.java b/src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandlerBase.java
index 5b1e9ada8..f1f2bf9ba 100644
--- a/src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandlerBase.java
+++ b/src/main/java/com/uber/cadence/internal/sync/ActivityInvocationHandlerBase.java
@@ -18,17 +18,19 @@
package com.uber.cadence.internal.sync;
import static com.uber.cadence.internal.common.InternalUtils.getValueOrDefault;
+import static com.uber.cadence.internal.sync.POJOActivityTaskHandler.getAnnotatedInterfaceMethodsFromInterface;
+import com.uber.cadence.activity.ActivityInterface;
import com.uber.cadence.activity.ActivityMethod;
import com.uber.cadence.common.MethodRetry;
import com.uber.cadence.internal.common.InternalUtils;
import com.uber.cadence.internal.sync.AsyncInternal.AsyncMarker;
-import com.uber.cadence.workflow.Workflow;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
import java.util.function.Function;
/** Dynamic implementation of a strongly typed child workflow interface. */
@@ -45,33 +47,36 @@ static T newProxy(Class activityInterface, InvocationHandler invocationHa
invocationHandler);
}
+ void init(Class activityInterface) {
+ Set activityMethods =
+ getAnnotatedInterfaceMethodsFromInterface(activityInterface, ActivityInterface.class);
+ if (activityMethods.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Class doesn't implement any non empty interface annotated with @ActivityInterface: "
+ + activityInterface.getName());
+ }
+ for (POJOActivityTaskHandler.MethodInterfacePair pair : activityMethods) {
+ Method method = pair.getMethod();
+ ActivityMethod activityMethod = method.getAnnotation(ActivityMethod.class);
+ String activityType;
+ if (activityMethod != null && !activityMethod.name().isEmpty()) {
+ activityType = activityMethod.name();
+ } else {
+ activityType = InternalUtils.getSimpleName(pair.getType(), method);
+ }
+
+ MethodRetry methodRetry = method.getAnnotation(MethodRetry.class);
+ Function function =
+ getActivityFunc(method, methodRetry, activityMethod, activityType);
+ methodFunctions.put(method, function);
+ }
+ }
+
@Override
public Object invoke(Object proxy, Method method, Object[] args) {
Function function = methodFunctions.get(method);
if (function == null) {
- try {
- if (method.equals(Object.class.getMethod("toString"))) {
- // TODO: activity info
- return "ActivityInvocationHandlerBase";
- }
- if (!method.getDeclaringClass().isInterface()) {
- throw new IllegalArgumentException(
- "Interface type is expected: " + method.getDeclaringClass());
- }
- MethodRetry methodRetry = method.getAnnotation(MethodRetry.class);
- ActivityMethod activityMethod = method.getAnnotation(ActivityMethod.class);
- String activityName;
- if (activityMethod == null || activityMethod.name().isEmpty()) {
- activityName = InternalUtils.getSimpleName(method);
- } else {
- activityName = activityMethod.name();
- }
-
- function = getActivityFunc(method, methodRetry, activityMethod, activityName);
- methodFunctions.put(method, function);
- } catch (NoSuchMethodException e) {
- throw Workflow.wrap(e);
- }
+ throw new IllegalArgumentException("Unexpected method: " + method);
}
return getValueOrDefault(function.apply(args), method.getReturnType());
}
diff --git a/src/main/java/com/uber/cadence/internal/sync/LocalActivityInvocationHandler.java b/src/main/java/com/uber/cadence/internal/sync/LocalActivityInvocationHandler.java
index 53888b81d..df4ebc085 100644
--- a/src/main/java/com/uber/cadence/internal/sync/LocalActivityInvocationHandler.java
+++ b/src/main/java/com/uber/cadence/internal/sync/LocalActivityInvocationHandler.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
@@ -24,6 +24,7 @@
import com.uber.cadence.workflow.WorkflowInterceptor;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
+import java.time.Duration;
import java.util.function.Function;
class LocalActivityInvocationHandler extends ActivityInvocationHandlerBase {
@@ -31,22 +32,36 @@ class LocalActivityInvocationHandler extends ActivityInvocationHandlerBase {
private final WorkflowInterceptor activityExecutor;
static InvocationHandler newInstance(
- LocalActivityOptions options, WorkflowInterceptor activityExecutor) {
- return new LocalActivityInvocationHandler(options, activityExecutor);
+ Class activityInterface,
+ LocalActivityOptions options,
+ WorkflowInterceptor activityExecutor) {
+ return new LocalActivityInvocationHandler(activityInterface, activityExecutor, options);
}
private LocalActivityInvocationHandler(
- LocalActivityOptions options, WorkflowInterceptor activityExecutor) {
+ Class activityInterface,
+ WorkflowInterceptor activityExecutor,
+ LocalActivityOptions options) {
this.options = options;
this.activityExecutor = activityExecutor;
+ init(activityInterface);
}
+ @SuppressWarnings("deprecation")
@Override
protected Function getActivityFunc(
Method method, MethodRetry methodRetry, ActivityMethod activityMethod, String activityName) {
Function function;
- LocalActivityOptions mergedOptions =
- LocalActivityOptions.merge(activityMethod, methodRetry, options);
+ LocalActivityOptions.Builder optionsBuilder =
+ LocalActivityOptions.newBuilder(options).setMethodRetry(methodRetry);
+ if (activityMethod != null) {
+ // options always take precedence over activity method annotation.
+ if (options.getScheduleToCloseTimeout() == null) {
+ optionsBuilder.setScheduleToCloseTimeout(
+ Duration.ofSeconds(activityMethod.scheduleToCloseTimeoutSeconds()));
+ }
+ }
+ LocalActivityOptions mergedOptions = optionsBuilder.validateAndBuildWithDefaults();
ActivityStub stub = LocalActivityStubImpl.newInstance(mergedOptions, activityExecutor);
function =
(a) -> stub.execute(activityName, method.getReturnType(), method.getGenericReturnType(), a);
diff --git a/src/main/java/com/uber/cadence/internal/sync/POJOActivityTaskHandler.java b/src/main/java/com/uber/cadence/internal/sync/POJOActivityTaskHandler.java
index ec49d1469..ba660b680 100644
--- a/src/main/java/com/uber/cadence/internal/sync/POJOActivityTaskHandler.java
+++ b/src/main/java/com/uber/cadence/internal/sync/POJOActivityTaskHandler.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
@@ -17,11 +17,13 @@
package com.uber.cadence.internal.sync;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
-import com.google.common.reflect.TypeToken;
+import com.google.common.base.Objects;
import com.uber.cadence.PollForActivityTaskResponse;
import com.uber.cadence.RespondActivityTaskCompletedRequest;
import com.uber.cadence.RespondActivityTaskFailedRequest;
+import com.uber.cadence.activity.ActivityInterface;
import com.uber.cadence.activity.ActivityMethod;
import com.uber.cadence.client.ActivityCancelledException;
import com.uber.cadence.common.MethodRetry;
@@ -33,11 +35,14 @@
import com.uber.cadence.serviceclient.IWorkflowService;
import com.uber.cadence.testing.SimulatedTimeoutException;
import com.uber.m3.tally.Scope;
+import java.lang.annotation.Annotation;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.BiFunction;
@@ -82,30 +87,29 @@ private void addActivityImplementation(
+ "\" This annotation can be used only on the interface method it implements.");
}
}
- TypeToken.TypeSet interfaces = TypeToken.of(cls).getTypes().interfaces();
- if (interfaces.isEmpty()) {
- throw new IllegalArgumentException("Activity must implement at least one interface");
+ Set activityMethods =
+ getAnnotatedInterfaceMethodsFromImplementation(cls, ActivityInterface.class);
+ if (activityMethods.isEmpty()) {
+ throw new IllegalArgumentException(
+ "Class doesn't implement any non empty interface annotated with @ActivityInterface: "
+ + cls.getName());
}
- for (TypeToken i : interfaces) {
- if (i.getType().getTypeName().startsWith("org.mockito")) {
- continue;
+ for (MethodInterfacePair pair : activityMethods) {
+ Method method = pair.getMethod();
+ ActivityMethod annotation = method.getAnnotation(ActivityMethod.class);
+ String activityType;
+ if (annotation != null && !annotation.name().isEmpty()) {
+ activityType = annotation.name();
+ } else {
+ activityType = InternalUtils.getSimpleName(pair.getType(), method);
}
- for (Method method : i.getRawType().getMethods()) {
- ActivityMethod annotation = method.getAnnotation(ActivityMethod.class);
- String activityType;
- if (annotation != null && !annotation.name().isEmpty()) {
- activityType = annotation.name();
- } else {
- activityType = InternalUtils.getSimpleName(method);
- }
- if (activities.containsKey(activityType)) {
- throw new IllegalStateException(
- activityType + " activity type is already registered with the worker");
- }
-
- ActivityTaskExecutor implementation = newTaskExecutor.apply(method, activity);
- activities.put(activityType, implementation);
+ if (activities.containsKey(activityType)) {
+ throw new IllegalStateException(
+ activityType + " activity type is already registered with the worker");
}
+
+ ActivityTaskExecutor implementation = newTaskExecutor.apply(method, activity);
+ activities.put(activityType, implementation);
}
}
@@ -147,8 +151,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
@@ -156,6 +159,11 @@ public boolean isAnyTypeSupported() {
return !activities.isEmpty();
}
+ @VisibleForTesting
+ public Set getRegisteredActivityTypes() {
+ return activities.keySet();
+ }
+
void setActivitiesImplementation(Object[] activitiesImplementation) {
activities.clear();
for (Object activity : activitiesImplementation) {
@@ -214,12 +222,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 +260,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) {
@@ -267,4 +275,131 @@ public ActivityTaskHandler.Result execute(ActivityTaskImpl task, Scope metricsSc
void setWorkflowService(IWorkflowService service) {
this.service = service;
}
+
+ static class MethodInterfacePair {
+ private final Method method;
+ private final Class type;
+
+ MethodInterfacePair(Method method, Class type) {
+ this.method = method;
+ this.type = type;
+ }
+
+ public Method getMethod() {
+ return method;
+ }
+
+ public Class getType() {
+ return type;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ MethodInterfacePair that = (MethodInterfacePair) o;
+ return Objects.equal(method, that.method) && Objects.equal(type, that.type);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(method, type);
+ }
+
+ @Override
+ public String toString() {
+ return "MethodInterfacePair{" + "method=" + method + ", type=" + type + '}';
+ }
+ }
+
+ /** Used to override equals and hashCode of Method to ensure deduping by method name in a set. */
+ static class MethodWrapper {
+ private final Method method;
+
+ MethodWrapper(Method method) {
+ this.method = method;
+ }
+
+ public Method getMethod() {
+ return method;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ MethodWrapper that = (MethodWrapper) o;
+ return Objects.equal(method.getName(), that.method.getName());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(method.getName());
+ }
+ }
+
+ static Set getAnnotatedInterfaceMethodsFromImplementation(
+ Class implementationClass, Class annotationClass) {
+ if (implementationClass.isInterface()) {
+ throw new IllegalArgumentException(
+ "Concrete class expected. Found interface: " + implementationClass.getSimpleName());
+ }
+ Set pairs = new HashSet<>();
+ // Methods inherited from interfaces that are not annotated with @ActivityInterface
+ Set ignored = new HashSet<>();
+ getAnnotatedInterfaceMethodsFromImplementation(
+ implementationClass, annotationClass, ignored, pairs);
+ return pairs;
+ }
+
+ static Set getAnnotatedInterfaceMethodsFromInterface(
+ Class iClass, Class annotationClass) {
+ if (!iClass.isInterface()) {
+ throw new IllegalArgumentException("Interface expected. Found: " + iClass.getSimpleName());
+ }
+ Annotation annotation = iClass.getAnnotation(annotationClass);
+ if (annotation == null) {
+ throw new IllegalArgumentException(
+ "@ActivityInterface annotation is required on the stub interface: "
+ + iClass.getSimpleName());
+ }
+ Set pairs = new HashSet<>();
+ // Methods inherited from interfaces that are not annotated with @ActivityInterface
+ Set ignored = new HashSet<>();
+ getAnnotatedInterfaceMethodsFromImplementation(iClass, annotationClass, ignored, pairs);
+ if (!ignored.isEmpty()) {
+ throw new IllegalStateException("Not empty ignored: " + ignored);
+ }
+ return pairs;
+ }
+
+ private static void getAnnotatedInterfaceMethodsFromImplementation(
+ Class current,
+ Class annotationClass,
+ Set methods,
+ Set result) {
+ // Using set to dedupe methods which are defined in both non activity parent and current
+ Set ourMethods = new HashSet<>();
+ if (current.isInterface()) {
+ Method[] declaredMethods = current.getDeclaredMethods();
+ for (int i = 0; i < declaredMethods.length; i++) {
+ Method declaredMethod = declaredMethods[i];
+ ourMethods.add(new MethodWrapper(declaredMethod));
+ }
+ }
+ Class[] interfaces = current.getInterfaces();
+ for (int i = 0; i < interfaces.length; i++) {
+ Class anInterface = interfaces[i];
+ getAnnotatedInterfaceMethodsFromImplementation(
+ anInterface, annotationClass, ourMethods, result);
+ }
+ Annotation annotation = current.getAnnotation(annotationClass);
+ if (annotation == null) {
+ methods.addAll(ourMethods);
+ return;
+ }
+ for (MethodWrapper method : ourMethods) {
+ result.add(new MethodInterfacePair(method.getMethod(), current));
+ }
+ }
}
diff --git a/src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.java b/src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.java
index 30dcb206d..95d065a8b 100644
--- a/src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.java
+++ b/src/main/java/com/uber/cadence/internal/sync/TestActivityEnvironmentInternal.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
@@ -18,7 +18,77 @@
package com.uber.cadence.internal.sync;
import com.google.common.base.Defaults;
-import com.uber.cadence.*;
+import com.uber.cadence.ActivityType;
+import com.uber.cadence.BadRequestError;
+import com.uber.cadence.CancellationAlreadyRequestedError;
+import com.uber.cadence.ClientVersionNotSupportedError;
+import com.uber.cadence.ClusterInfo;
+import com.uber.cadence.CountWorkflowExecutionsRequest;
+import com.uber.cadence.CountWorkflowExecutionsResponse;
+import com.uber.cadence.DeprecateDomainRequest;
+import com.uber.cadence.DescribeDomainRequest;
+import com.uber.cadence.DescribeDomainResponse;
+import com.uber.cadence.DescribeTaskListRequest;
+import com.uber.cadence.DescribeTaskListResponse;
+import com.uber.cadence.DescribeWorkflowExecutionRequest;
+import com.uber.cadence.DescribeWorkflowExecutionResponse;
+import com.uber.cadence.DomainAlreadyExistsError;
+import com.uber.cadence.DomainNotActiveError;
+import com.uber.cadence.EntityNotExistsError;
+import com.uber.cadence.GetSearchAttributesResponse;
+import com.uber.cadence.GetWorkflowExecutionHistoryRequest;
+import com.uber.cadence.GetWorkflowExecutionHistoryResponse;
+import com.uber.cadence.InternalServiceError;
+import com.uber.cadence.LimitExceededError;
+import com.uber.cadence.ListArchivedWorkflowExecutionsRequest;
+import com.uber.cadence.ListArchivedWorkflowExecutionsResponse;
+import com.uber.cadence.ListClosedWorkflowExecutionsRequest;
+import com.uber.cadence.ListClosedWorkflowExecutionsResponse;
+import com.uber.cadence.ListDomainsRequest;
+import com.uber.cadence.ListDomainsResponse;
+import com.uber.cadence.ListOpenWorkflowExecutionsRequest;
+import com.uber.cadence.ListOpenWorkflowExecutionsResponse;
+import com.uber.cadence.ListTaskListPartitionsRequest;
+import com.uber.cadence.ListTaskListPartitionsResponse;
+import com.uber.cadence.ListWorkflowExecutionsRequest;
+import com.uber.cadence.ListWorkflowExecutionsResponse;
+import com.uber.cadence.PollForActivityTaskRequest;
+import com.uber.cadence.PollForActivityTaskResponse;
+import com.uber.cadence.PollForDecisionTaskRequest;
+import com.uber.cadence.PollForDecisionTaskResponse;
+import com.uber.cadence.QueryFailedError;
+import com.uber.cadence.QueryWorkflowRequest;
+import com.uber.cadence.QueryWorkflowResponse;
+import com.uber.cadence.RecordActivityTaskHeartbeatByIDRequest;
+import com.uber.cadence.RecordActivityTaskHeartbeatRequest;
+import com.uber.cadence.RecordActivityTaskHeartbeatResponse;
+import com.uber.cadence.RegisterDomainRequest;
+import com.uber.cadence.RequestCancelWorkflowExecutionRequest;
+import com.uber.cadence.ResetStickyTaskListRequest;
+import com.uber.cadence.ResetStickyTaskListResponse;
+import com.uber.cadence.ResetWorkflowExecutionRequest;
+import com.uber.cadence.ResetWorkflowExecutionResponse;
+import com.uber.cadence.RespondActivityTaskCanceledByIDRequest;
+import com.uber.cadence.RespondActivityTaskCanceledRequest;
+import com.uber.cadence.RespondActivityTaskCompletedByIDRequest;
+import com.uber.cadence.RespondActivityTaskCompletedRequest;
+import com.uber.cadence.RespondActivityTaskFailedByIDRequest;
+import com.uber.cadence.RespondActivityTaskFailedRequest;
+import com.uber.cadence.RespondDecisionTaskCompletedRequest;
+import com.uber.cadence.RespondDecisionTaskCompletedResponse;
+import com.uber.cadence.RespondDecisionTaskFailedRequest;
+import com.uber.cadence.RespondQueryTaskCompletedRequest;
+import com.uber.cadence.ServiceBusyError;
+import com.uber.cadence.SignalWithStartWorkflowExecutionRequest;
+import com.uber.cadence.SignalWorkflowExecutionRequest;
+import com.uber.cadence.StartWorkflowExecutionRequest;
+import com.uber.cadence.StartWorkflowExecutionResponse;
+import com.uber.cadence.TerminateWorkflowExecutionRequest;
+import com.uber.cadence.UpdateDomainRequest;
+import com.uber.cadence.UpdateDomainResponse;
+import com.uber.cadence.WorkflowExecution;
+import com.uber.cadence.WorkflowExecutionAlreadyStartedError;
+import com.uber.cadence.WorkflowType;
import com.uber.cadence.activity.ActivityOptions;
import com.uber.cadence.activity.LocalActivityOptions;
import com.uber.cadence.internal.metrics.NoopScope;
@@ -102,9 +172,13 @@ public void registerActivitiesImplementations(Object... activityImplementations)
@Override
public T newActivityStub(Class activityInterface) {
ActivityOptions options =
- new ActivityOptions.Builder().setScheduleToCloseTimeout(Duration.ofDays(1)).build();
+ ActivityOptions.newBuilder()
+ .setScheduleToCloseTimeout(Duration.ofDays(1))
+ .setHeartbeatTimeout(Duration.ofSeconds(1))
+ .build();
InvocationHandler invocationHandler =
- ActivityInvocationHandler.newInstance(options, new TestActivityExecutor(workflowService));
+ ActivityInvocationHandler.newInstance(
+ activityInterface, options, new TestActivityExecutor(workflowService));
invocationHandler = new DeterministicRunnerWrapper(invocationHandler);
return ActivityInvocationHandlerBase.newProxy(activityInterface, invocationHandler);
}
diff --git a/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java b/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java
index 4e61afc6f..236aa3bca 100644
--- a/src/main/java/com/uber/cadence/internal/sync/TestWorkflowEnvironmentInternal.java
+++ b/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());
}
diff --git a/src/main/java/com/uber/cadence/internal/sync/WorkflowInternal.java b/src/main/java/com/uber/cadence/internal/sync/WorkflowInternal.java
index 9ada6ad1f..6140e4de0 100644
--- a/src/main/java/com/uber/cadence/internal/sync/WorkflowInternal.java
+++ b/src/main/java/com/uber/cadence/internal/sync/WorkflowInternal.java
@@ -149,7 +149,7 @@ public static long currentTimeMillis() {
public static T newActivityStub(Class activityInterface, ActivityOptions options) {
WorkflowInterceptor decisionContext = WorkflowInternal.getWorkflowInterceptor();
InvocationHandler invocationHandler =
- ActivityInvocationHandler.newInstance(options, decisionContext);
+ ActivityInvocationHandler.newInstance(activityInterface, options, decisionContext);
return ActivityInvocationHandlerBase.newProxy(activityInterface, invocationHandler);
}
@@ -162,7 +162,7 @@ public static T newLocalActivityStub(
Class activityInterface, LocalActivityOptions options) {
WorkflowInterceptor decisionContext = WorkflowInternal.getWorkflowInterceptor();
InvocationHandler invocationHandler =
- LocalActivityInvocationHandler.newInstance(options, decisionContext);
+ LocalActivityInvocationHandler.newInstance(activityInterface, options, decisionContext);
return ActivityInvocationHandlerBase.newProxy(activityInterface, invocationHandler);
}
diff --git a/src/main/java/com/uber/cadence/internal/worker/ActivityTaskHandler.java b/src/main/java/com/uber/cadence/internal/worker/ActivityTaskHandler.java
index e49e5ff56..9d18c1ad9 100644
--- a/src/main/java/com/uber/cadence/internal/worker/ActivityTaskHandler.java
+++ b/src/main/java/com/uber/cadence/internal/worker/ActivityTaskHandler.java
@@ -21,7 +21,6 @@
import com.uber.cadence.RespondActivityTaskCanceledRequest;
import com.uber.cadence.RespondActivityTaskCompletedRequest;
import com.uber.cadence.RespondActivityTaskFailedRequest;
-import com.uber.cadence.common.RetryOptions;
import com.uber.m3.tally.Scope;
import java.time.Duration;
@@ -37,7 +36,6 @@ final class Result {
private final RespondActivityTaskCompletedRequest taskCompleted;
private final TaskFailedResult taskFailed;
private final RespondActivityTaskCanceledRequest taskCancelled;
- private final RetryOptions requestRetryOptions;
private int attempt;
private Duration backoff;
@@ -68,12 +66,10 @@ public Throwable getFailure() {
public Result(
RespondActivityTaskCompletedRequest taskCompleted,
TaskFailedResult taskFailed,
- RespondActivityTaskCanceledRequest taskCancelled,
- RetryOptions requestRetryOptions) {
+ RespondActivityTaskCanceledRequest taskCancelled) {
this.taskCompleted = taskCompleted;
this.taskFailed = taskFailed;
this.taskCancelled = taskCancelled;
- this.requestRetryOptions = requestRetryOptions;
}
public RespondActivityTaskCompletedRequest getTaskCompleted() {
@@ -88,10 +84,6 @@ public RespondActivityTaskCanceledRequest getTaskCancelled() {
return taskCancelled;
}
- public RetryOptions getRequestRetryOptions() {
- return requestRetryOptions;
- }
-
public void setAttempt(int attempt) {
this.attempt = attempt;
}
diff --git a/src/main/java/com/uber/cadence/internal/worker/ActivityWorker.java b/src/main/java/com/uber/cadence/internal/worker/ActivityWorker.java
index 01d65383b..a0a86014a 100644
--- a/src/main/java/com/uber/cadence/internal/worker/ActivityWorker.java
+++ b/src/main/java/com/uber/cadence/internal/worker/ActivityWorker.java
@@ -17,18 +17,14 @@
package com.uber.cadence.internal.worker;
-import com.uber.cadence.BadRequestError;
-import com.uber.cadence.DomainNotActiveError;
-import com.uber.cadence.EntityNotExistsError;
import com.uber.cadence.Header;
import com.uber.cadence.PollForActivityTaskResponse;
import com.uber.cadence.RespondActivityTaskCanceledRequest;
import com.uber.cadence.RespondActivityTaskCompletedRequest;
import com.uber.cadence.RespondActivityTaskFailedRequest;
import com.uber.cadence.WorkflowExecution;
-import com.uber.cadence.common.RetryOptions;
import com.uber.cadence.context.ContextPropagator;
-import com.uber.cadence.internal.common.Retryer;
+import com.uber.cadence.internal.common.RpcRetryer;
import com.uber.cadence.internal.logging.LoggerTag;
import com.uber.cadence.internal.metrics.MetricsTag;
import com.uber.cadence.internal.metrics.MetricsType;
@@ -150,7 +146,7 @@ public void handle(PollForActivityTaskResponse task) throws Exception {
cancelledRequest.setDetails(
String.valueOf(e.getMessage()).getBytes(StandardCharsets.UTF_8));
Stopwatch sw = metricsScope.timer(MetricsType.ACTIVITY_RESP_LATENCY).start();
- sendReply(task, new Result(null, null, cancelledRequest, null), metricsScope);
+ sendReply(task, new Result(null, null, cancelledRequest), metricsScope);
sw.stop();
} finally {
MDC.remove(LoggerTag.ACTIVITY_ID);
@@ -201,49 +197,26 @@ public Throwable wrapFailure(PollForActivityTaskResponse task, Throwable failure
private void sendReply(
PollForActivityTaskResponse task, ActivityTaskHandler.Result response, Scope metricsScope)
throws TException {
- RetryOptions ro = response.getRequestRetryOptions();
RespondActivityTaskCompletedRequest taskCompleted = response.getTaskCompleted();
if (taskCompleted != null) {
- ro =
- options
- .getReportCompletionRetryOptions()
- .merge(ro)
- .addDoNotRetry(
- BadRequestError.class, EntityNotExistsError.class, DomainNotActiveError.class);
taskCompleted.setTaskToken(task.getTaskToken());
taskCompleted.setIdentity(options.getIdentity());
- Retryer.retry(ro, () -> service.RespondActivityTaskCompleted(taskCompleted));
+ RpcRetryer.retry(() -> service.RespondActivityTaskCompleted(taskCompleted));
metricsScope.counter(MetricsType.ACTIVITY_TASK_COMPLETED_COUNTER).inc(1);
} else {
if (response.getTaskFailedResult() != null) {
RespondActivityTaskFailedRequest taskFailed =
response.getTaskFailedResult().getTaskFailedRequest();
- ro =
- options
- .getReportFailureRetryOptions()
- .merge(ro)
- .addDoNotRetry(
- BadRequestError.class,
- EntityNotExistsError.class,
- DomainNotActiveError.class);
taskFailed.setTaskToken(task.getTaskToken());
taskFailed.setIdentity(options.getIdentity());
- Retryer.retry(ro, () -> service.RespondActivityTaskFailed(taskFailed));
+ RpcRetryer.retry(() -> service.RespondActivityTaskFailed(taskFailed));
metricsScope.counter(MetricsType.ACTIVITY_TASK_FAILED_COUNTER).inc(1);
} else {
RespondActivityTaskCanceledRequest taskCancelled = response.getTaskCancelled();
if (taskCancelled != null) {
taskCancelled.setTaskToken(task.getTaskToken());
taskCancelled.setIdentity(options.getIdentity());
- ro =
- options
- .getReportFailureRetryOptions()
- .merge(ro)
- .addDoNotRetry(
- BadRequestError.class,
- EntityNotExistsError.class,
- DomainNotActiveError.class);
- Retryer.retry(ro, () -> service.RespondActivityTaskCanceled(taskCancelled));
+ RpcRetryer.retry(() -> service.RespondActivityTaskCanceled(taskCancelled));
metricsScope.counter(MetricsType.ACTIVITY_TASK_CANCELED_COUNTER).inc(1);
}
}
diff --git a/src/main/java/com/uber/cadence/internal/worker/DecisionTaskHandler.java b/src/main/java/com/uber/cadence/internal/worker/DecisionTaskHandler.java
index 2193e82d1..70c10f55f 100644
--- a/src/main/java/com/uber/cadence/internal/worker/DecisionTaskHandler.java
+++ b/src/main/java/com/uber/cadence/internal/worker/DecisionTaskHandler.java
@@ -21,7 +21,6 @@
import com.uber.cadence.RespondDecisionTaskCompletedRequest;
import com.uber.cadence.RespondDecisionTaskFailedRequest;
import com.uber.cadence.RespondQueryTaskCompletedRequest;
-import com.uber.cadence.common.RetryOptions;
/**
* Interface of workflow task handlers.
@@ -34,17 +33,14 @@ final class Result {
private final RespondDecisionTaskCompletedRequest taskCompleted;
private final RespondDecisionTaskFailedRequest taskFailed;
private final RespondQueryTaskCompletedRequest queryCompleted;
- private final RetryOptions requestRetryOptions;
public Result(
RespondDecisionTaskCompletedRequest taskCompleted,
RespondDecisionTaskFailedRequest taskFailed,
- RespondQueryTaskCompletedRequest queryCompleted,
- RetryOptions requestRetryOptions) {
+ RespondQueryTaskCompletedRequest queryCompleted) {
this.taskCompleted = taskCompleted;
this.taskFailed = taskFailed;
this.queryCompleted = queryCompleted;
- this.requestRetryOptions = requestRetryOptions;
}
public RespondDecisionTaskCompletedRequest getTaskCompleted() {
@@ -59,10 +55,6 @@ public RespondQueryTaskCompletedRequest getQueryCompleted() {
return queryCompleted;
}
- public RetryOptions getRequestRetryOptions() {
- return requestRetryOptions;
- }
-
@Override
public String toString() {
return "Result{"
@@ -72,8 +64,6 @@ public String toString() {
+ taskFailed
+ ", queryCompleted="
+ queryCompleted
- + ", requestRetryOptions="
- + requestRetryOptions
+ '}';
}
}
diff --git a/src/main/java/com/uber/cadence/internal/worker/PollerOptions.java b/src/main/java/com/uber/cadence/internal/worker/PollerOptions.java
index 9b78546b1..23d43feea 100644
--- a/src/main/java/com/uber/cadence/internal/worker/PollerOptions.java
+++ b/src/main/java/com/uber/cadence/internal/worker/PollerOptions.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
diff --git a/src/main/java/com/uber/cadence/internal/worker/SingleWorkerOptions.java b/src/main/java/com/uber/cadence/internal/worker/SingleWorkerOptions.java
index 3a6d66916..3216a66e0 100644
--- a/src/main/java/com/uber/cadence/internal/worker/SingleWorkerOptions.java
+++ b/src/main/java/com/uber/cadence/internal/worker/SingleWorkerOptions.java
@@ -17,11 +17,9 @@
package com.uber.cadence.internal.worker;
-import com.uber.cadence.common.RetryOptions;
import com.uber.cadence.context.ContextPropagator;
import com.uber.cadence.converter.DataConverter;
import com.uber.cadence.converter.JsonDataConverter;
-import com.uber.cadence.internal.common.Retryer;
import com.uber.cadence.internal.metrics.NoopScope;
import com.uber.m3.tally.Scope;
import java.time.Duration;
@@ -43,10 +41,6 @@ public static final class Builder {
private int taskExecutorThreadPoolSize = 100;
private double taskListActivitiesPerSecond;
private PollerOptions pollerOptions;
- /** TODO: Dynamic expiration based on activity timeout */
- private RetryOptions reportCompletionRetryOptions;
-
- private RetryOptions reportFailureRetryOptions;
private Scope metricsScope;
private boolean enableLoggingInReplay;
private List contextPropagators;
@@ -59,8 +53,6 @@ public Builder(SingleWorkerOptions options) {
this.pollerOptions = options.getPollerOptions();
this.taskListActivitiesPerSecond = options.getTaskListActivitiesPerSecond();
this.taskExecutorThreadPoolSize = options.getTaskExecutorThreadPoolSize();
- this.reportCompletionRetryOptions = options.getReportCompletionRetryOptions();
- this.reportFailureRetryOptions = options.getReportFailureRetryOptions();
this.metricsScope = options.getMetricsScope();
this.enableLoggingInReplay = options.getEnableLoggingInReplay();
this.contextPropagators = options.getContextPropagators();
@@ -101,16 +93,6 @@ public Builder setTaskListActivitiesPerSecond(double taskListActivitiesPerSecond
return this;
}
- public Builder setReportCompletionRetryOptions(RetryOptions reportCompletionRetryOptions) {
- this.reportCompletionRetryOptions = reportCompletionRetryOptions;
- return this;
- }
-
- public Builder setReportFailureRetryOptions(RetryOptions reportFailureRetryOptions) {
- this.reportFailureRetryOptions = reportFailureRetryOptions;
- return this;
- }
-
/** Specifies the list of context propagators to use during this workflow. */
public Builder setContextPropagators(List contextPropagators) {
this.contextPropagators = contextPropagators;
@@ -118,14 +100,6 @@ public Builder setContextPropagators(List contextPropagators)
}
public SingleWorkerOptions build() {
- if (reportCompletionRetryOptions == null) {
- reportCompletionRetryOptions = Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS;
- }
-
- if (reportFailureRetryOptions == null) {
- reportFailureRetryOptions = Retryer.DEFAULT_SERVICE_OPERATION_RETRY_OPTIONS;
- }
-
if (pollerOptions == null) {
pollerOptions =
PollerOptions.newBuilder()
@@ -149,8 +123,6 @@ public SingleWorkerOptions build() {
taskExecutorThreadPoolSize,
taskListActivitiesPerSecond,
pollerOptions,
- reportCompletionRetryOptions,
- reportFailureRetryOptions,
metricsScope,
enableLoggingInReplay,
contextPropagators);
@@ -162,8 +134,6 @@ public SingleWorkerOptions build() {
private final int taskExecutorThreadPoolSize;
private final double taskListActivitiesPerSecond;
private final PollerOptions pollerOptions;
- private final RetryOptions reportCompletionRetryOptions;
- private final RetryOptions reportFailureRetryOptions;
private final Scope metricsScope;
private final boolean enableLoggingInReplay;
private List contextPropagators;
@@ -174,8 +144,6 @@ private SingleWorkerOptions(
int taskExecutorThreadPoolSize,
double taskListActivitiesPerSecond,
PollerOptions pollerOptions,
- RetryOptions reportCompletionRetryOptions,
- RetryOptions reportFailureRetryOptions,
Scope metricsScope,
boolean enableLoggingInReplay,
List contextPropagators) {
@@ -184,8 +152,6 @@ private SingleWorkerOptions(
this.taskExecutorThreadPoolSize = taskExecutorThreadPoolSize;
this.taskListActivitiesPerSecond = taskListActivitiesPerSecond;
this.pollerOptions = pollerOptions;
- this.reportCompletionRetryOptions = reportCompletionRetryOptions;
- this.reportFailureRetryOptions = reportFailureRetryOptions;
this.metricsScope = metricsScope;
this.enableLoggingInReplay = enableLoggingInReplay;
this.contextPropagators = contextPropagators;
@@ -207,14 +173,6 @@ PollerOptions getPollerOptions() {
return pollerOptions;
}
- RetryOptions getReportCompletionRetryOptions() {
- return reportCompletionRetryOptions;
- }
-
- RetryOptions getReportFailureRetryOptions() {
- return reportFailureRetryOptions;
- }
-
double getTaskListActivitiesPerSecond() {
return taskListActivitiesPerSecond;
}
diff --git a/src/main/java/com/uber/cadence/internal/worker/WorkflowWorker.java b/src/main/java/com/uber/cadence/internal/worker/WorkflowWorker.java
index 959024ea3..d6a7bbeee 100644
--- a/src/main/java/com/uber/cadence/internal/worker/WorkflowWorker.java
+++ b/src/main/java/com/uber/cadence/internal/worker/WorkflowWorker.java
@@ -18,9 +18,6 @@
package com.uber.cadence.internal.worker;
import com.google.common.base.Strings;
-import com.uber.cadence.BadRequestError;
-import com.uber.cadence.DomainNotActiveError;
-import com.uber.cadence.EntityNotExistsError;
import com.uber.cadence.GetWorkflowExecutionHistoryResponse;
import com.uber.cadence.History;
import com.uber.cadence.HistoryEvent;
@@ -32,9 +29,8 @@
import com.uber.cadence.WorkflowExecutionStartedEventAttributes;
import com.uber.cadence.WorkflowQuery;
import com.uber.cadence.WorkflowType;
-import com.uber.cadence.common.RetryOptions;
import com.uber.cadence.common.WorkflowExecutionHistory;
-import com.uber.cadence.internal.common.Retryer;
+import com.uber.cadence.internal.common.RpcRetryer;
import com.uber.cadence.internal.common.WorkflowExecutionUtils;
import com.uber.cadence.internal.logging.LoggerTag;
import com.uber.cadence.internal.metrics.MetricsTag;
@@ -237,32 +233,17 @@ public Throwable wrapFailure(PollForDecisionTaskResponse task, Throwable failure
private void sendReply(
IWorkflowService service, byte[] taskToken, DecisionTaskHandler.Result response)
throws TException {
- RetryOptions ro = response.getRequestRetryOptions();
RespondDecisionTaskCompletedRequest taskCompleted = response.getTaskCompleted();
if (taskCompleted != null) {
- ro =
- options
- .getReportCompletionRetryOptions()
- .merge(ro)
- .addDoNotRetry(
- BadRequestError.class, EntityNotExistsError.class, DomainNotActiveError.class);
taskCompleted.setIdentity(options.getIdentity());
taskCompleted.setTaskToken(taskToken);
- Retryer.retry(ro, () -> service.RespondDecisionTaskCompleted(taskCompleted));
+ RpcRetryer.retry(() -> service.RespondDecisionTaskCompleted(taskCompleted));
} else {
RespondDecisionTaskFailedRequest taskFailed = response.getTaskFailed();
if (taskFailed != null) {
- ro =
- options
- .getReportFailureRetryOptions()
- .merge(ro)
- .addDoNotRetry(
- BadRequestError.class,
- EntityNotExistsError.class,
- DomainNotActiveError.class);
taskFailed.setIdentity(options.getIdentity());
taskFailed.setTaskToken(taskToken);
- Retryer.retry(ro, () -> service.RespondDecisionTaskFailed(taskFailed));
+ RpcRetryer.retry(() -> service.RespondDecisionTaskFailed(taskFailed));
} else {
RespondQueryTaskCompletedRequest queryCompleted = response.getQueryCompleted();
if (queryCompleted != null) {
diff --git a/src/main/java/com/uber/cadence/serviceclient/ClientOptions.java b/src/main/java/com/uber/cadence/serviceclient/ClientOptions.java
index b8781f861..3cb9b08ce 100644
--- a/src/main/java/com/uber/cadence/serviceclient/ClientOptions.java
+++ b/src/main/java/com/uber/cadence/serviceclient/ClientOptions.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-2020 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
diff --git a/src/main/java/com/uber/cadence/worker/Worker.java b/src/main/java/com/uber/cadence/worker/Worker.java
index 6fd167988..1dd9eb79e 100644
--- a/src/main/java/com/uber/cadence/worker/Worker.java
+++ b/src/main/java/com/uber/cadence/worker/Worker.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
@@ -84,11 +84,9 @@ public final class Worker implements Suspendable {
SingleWorkerOptions activityOptions =
SingleWorkerOptions.newBuilder()
- .setDataConverter(options.getDataConverter())
+ .setDataConverter(client.getOptions().getDataConverter())
.setIdentity(client.getOptions().getIdentity())
.setPollerOptions(options.getActivityPollerOptions())
- .setReportCompletionRetryOptions(options.getReportActivityCompletionRetryOptions())
- .setReportFailureRetryOptions(options.getReportActivityFailureRetryOptions())
.setTaskExecutorThreadPoolSize(options.getMaxConcurrentActivityExecutionSize())
.setMetricsScope(metricsScope)
.setEnableLoggingInReplay(factoryOptions.isEnableLoggingInReplay())
@@ -100,11 +98,9 @@ public final class Worker implements Suspendable {
SingleWorkerOptions workflowOptions =
SingleWorkerOptions.newBuilder()
- .setDataConverter(options.getDataConverter())
+ .setDataConverter(client.getOptions().getDataConverter())
.setIdentity(client.getOptions().getIdentity())
.setPollerOptions(options.getWorkflowPollerOptions())
- .setReportCompletionRetryOptions(options.getReportWorkflowCompletionRetryOptions())
- .setReportFailureRetryOptions(options.getReportWorkflowFailureRetryOptions())
.setTaskExecutorThreadPoolSize(options.getMaxConcurrentWorkflowExecutionSize())
.setMetricsScope(metricsScope)
.setEnableLoggingInReplay(factoryOptions.isEnableLoggingInReplay())
@@ -112,11 +108,9 @@ public final class Worker implements Suspendable {
.build();
SingleWorkerOptions localActivityOptions =
SingleWorkerOptions.newBuilder()
- .setDataConverter(options.getDataConverter())
+ .setDataConverter(client.getOptions().getDataConverter())
.setIdentity(client.getOptions().getIdentity())
.setPollerOptions(options.getWorkflowPollerOptions())
- .setReportCompletionRetryOptions(options.getReportWorkflowCompletionRetryOptions())
- .setReportFailureRetryOptions(options.getReportWorkflowFailureRetryOptions())
.setTaskExecutorThreadPoolSize(options.getMaxConcurrentLocalActivityExecutionSize())
.setMetricsScope(metricsScope)
.setEnableLoggingInReplay(factoryOptions.isEnableLoggingInReplay())
diff --git a/src/main/java/com/uber/cadence/worker/WorkerFactoryOptions.java b/src/main/java/com/uber/cadence/worker/WorkerFactoryOptions.java
index 5eb876d68..6d197a0c2 100644
--- a/src/main/java/com/uber/cadence/worker/WorkerFactoryOptions.java
+++ b/src/main/java/com/uber/cadence/worker/WorkerFactoryOptions.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
diff --git a/src/main/java/com/uber/cadence/worker/WorkerOptions.java b/src/main/java/com/uber/cadence/worker/WorkerOptions.java
index 799b81e94..cb66b3ade 100644
--- a/src/main/java/com/uber/cadence/worker/WorkerOptions.java
+++ b/src/main/java/com/uber/cadence/worker/WorkerOptions.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
@@ -17,9 +17,6 @@
package com.uber.cadence.worker;
-import com.uber.cadence.common.RetryOptions;
-import com.uber.cadence.converter.DataConverter;
-import com.uber.cadence.converter.JsonDataConverter;
import com.uber.cadence.internal.worker.PollerOptions;
import com.uber.cadence.workflow.WorkflowInterceptor;
import java.util.Objects;
@@ -47,24 +44,18 @@ public static WorkerOptions defaultInstance() {
public static final class Builder {
private double workerActivitiesPerSecond;
- private DataConverter dataConverter = JsonDataConverter.getInstance();
private int maxConcurrentActivityExecutionSize = 100;
private int maxConcurrentWorkflowExecutionSize = 50;
private int maxConcurrentLocalActivityExecutionSize = 100;
private double taskListActivitiesPerSecond = 100000;
private PollerOptions activityPollerOptions;
private PollerOptions workflowPollerOptions;
- private RetryOptions reportActivityCompletionRetryOptions;
- private RetryOptions reportActivityFailureRetryOptions;
- private RetryOptions reportWorkflowCompletionRetryOptions;
- private RetryOptions reportWorkflowFailureRetryOptions;
private Function interceptorFactory = (n) -> n;
private Builder() {}
private Builder(WorkerOptions options) {
this.workerActivitiesPerSecond = options.workerActivitiesPerSecond;
- this.dataConverter = options.dataConverter;
this.maxConcurrentActivityExecutionSize = options.maxConcurrentActivityExecutionSize;
this.maxConcurrentWorkflowExecutionSize = options.maxConcurrentWorkflowExecutionSize;
this.maxConcurrentLocalActivityExecutionSize =
@@ -72,22 +63,9 @@ private Builder(WorkerOptions options) {
this.taskListActivitiesPerSecond = options.taskListActivitiesPerSecond;
this.activityPollerOptions = options.activityPollerOptions;
this.workflowPollerOptions = options.workflowPollerOptions;
- this.reportActivityCompletionRetryOptions = options.reportActivityCompletionRetryOptions;
- this.reportActivityFailureRetryOptions = options.reportActivityFailureRetryOptions;
- this.reportWorkflowCompletionRetryOptions = options.reportWorkflowCompletionRetryOptions;
- this.reportWorkflowFailureRetryOptions = options.reportWorkflowFailureRetryOptions;
this.interceptorFactory = options.interceptorFactory;
}
- /**
- * Override a data converter implementation used by workflows and activities executed by this
- * worker. Default is {@link com.uber.cadence.converter.JsonDataConverter} data converter.
- */
- public Builder setDataConverter(DataConverter dataConverter) {
- this.dataConverter = Objects.requireNonNull(dataConverter);
- return this;
- }
-
/** Maximum number of activities started per second. Default is 0 which means unlimited. */
public Builder setWorkerActivitiesPerSecond(double workerActivitiesPerSecond) {
if (workerActivitiesPerSecond <= 0) {
@@ -138,34 +116,6 @@ public Builder setWorkflowPollerOptions(PollerOptions workflowPollerOptions) {
return this;
}
- public Builder setReportActivityCompletionRetryOptions(
- RetryOptions reportActivityCompletionRetryOptions) {
- this.reportActivityCompletionRetryOptions =
- Objects.requireNonNull(reportActivityCompletionRetryOptions);
- return this;
- }
-
- public Builder setReportActivityFailureRetryOptions(
- RetryOptions reportActivityFailureRetryOptions) {
- this.reportActivityFailureRetryOptions =
- Objects.requireNonNull(reportActivityFailureRetryOptions);
- return this;
- }
-
- public Builder setReportWorkflowCompletionRetryOptions(
- RetryOptions reportWorkflowCompletionRetryOptions) {
- this.reportWorkflowCompletionRetryOptions =
- Objects.requireNonNull(reportWorkflowCompletionRetryOptions);
- return this;
- }
-
- public Builder setReportWorkflowFailureRetryOptions(
- RetryOptions reportWorkflowFailureRetryOptions) {
- this.reportWorkflowFailureRetryOptions =
- Objects.requireNonNull(reportWorkflowFailureRetryOptions);
- return this;
- }
-
public Builder setInterceptorFactory(
Function interceptorFactory) {
this.interceptorFactory = Objects.requireNonNull(interceptorFactory);
@@ -188,61 +138,41 @@ public Builder setTaskListActivitiesPerSecond(double taskListActivitiesPerSecond
public WorkerOptions build() {
return new WorkerOptions(
workerActivitiesPerSecond,
- dataConverter,
maxConcurrentActivityExecutionSize,
maxConcurrentWorkflowExecutionSize,
maxConcurrentLocalActivityExecutionSize,
taskListActivitiesPerSecond,
activityPollerOptions,
workflowPollerOptions,
- reportActivityCompletionRetryOptions,
- reportActivityFailureRetryOptions,
- reportWorkflowCompletionRetryOptions,
- reportWorkflowFailureRetryOptions,
interceptorFactory);
}
}
private final double workerActivitiesPerSecond;
- private final DataConverter dataConverter;
private final int maxConcurrentActivityExecutionSize;
private final int maxConcurrentWorkflowExecutionSize;
private final int maxConcurrentLocalActivityExecutionSize;
private final double taskListActivitiesPerSecond;
private final PollerOptions activityPollerOptions;
private final PollerOptions workflowPollerOptions;
- private final RetryOptions reportActivityCompletionRetryOptions;
- private final RetryOptions reportActivityFailureRetryOptions;
- private final RetryOptions reportWorkflowCompletionRetryOptions;
- private final RetryOptions reportWorkflowFailureRetryOptions;
private final Function interceptorFactory;
private WorkerOptions(
double workerActivitiesPerSecond,
- DataConverter dataConverter,
int maxConcurrentActivityExecutionSize,
int maxConcurrentWorkflowExecutionSize,
int maxConcurrentLocalActivityExecutionSize,
double taskListActivitiesPerSecond,
PollerOptions activityPollerOptions,
PollerOptions workflowPollerOptions,
- RetryOptions reportActivityCompletionRetryOptions,
- RetryOptions reportActivityFailureRetryOptions,
- RetryOptions reportWorkflowCompletionRetryOptions,
- RetryOptions reportWorkflowFailureRetryOptions,
Function interceptorFactory) {
this.workerActivitiesPerSecond = workerActivitiesPerSecond;
- this.dataConverter = dataConverter;
this.maxConcurrentActivityExecutionSize = maxConcurrentActivityExecutionSize;
this.maxConcurrentWorkflowExecutionSize = maxConcurrentWorkflowExecutionSize;
this.maxConcurrentLocalActivityExecutionSize = maxConcurrentLocalActivityExecutionSize;
this.taskListActivitiesPerSecond = taskListActivitiesPerSecond;
this.activityPollerOptions = activityPollerOptions;
this.workflowPollerOptions = workflowPollerOptions;
- this.reportActivityCompletionRetryOptions = reportActivityCompletionRetryOptions;
- this.reportActivityFailureRetryOptions = reportActivityFailureRetryOptions;
- this.reportWorkflowCompletionRetryOptions = reportWorkflowCompletionRetryOptions;
- this.reportWorkflowFailureRetryOptions = reportWorkflowFailureRetryOptions;
this.interceptorFactory = interceptorFactory;
}
@@ -250,10 +180,6 @@ public double getWorkerActivitiesPerSecond() {
return workerActivitiesPerSecond;
}
- public DataConverter getDataConverter() {
- return dataConverter;
- }
-
public int getMaxConcurrentActivityExecutionSize() {
return maxConcurrentActivityExecutionSize;
}
@@ -274,22 +200,6 @@ public PollerOptions getWorkflowPollerOptions() {
return workflowPollerOptions;
}
- public RetryOptions getReportActivityCompletionRetryOptions() {
- return reportActivityCompletionRetryOptions;
- }
-
- public RetryOptions getReportActivityFailureRetryOptions() {
- return reportActivityFailureRetryOptions;
- }
-
- public RetryOptions getReportWorkflowCompletionRetryOptions() {
- return reportWorkflowCompletionRetryOptions;
- }
-
- public RetryOptions getReportWorkflowFailureRetryOptions() {
- return reportWorkflowFailureRetryOptions;
- }
-
public Function getInterceptorFactory() {
return interceptorFactory;
}
@@ -299,8 +209,6 @@ public String toString() {
return "WorkerOptions{"
+ "workerActivitiesPerSecond="
+ workerActivitiesPerSecond
- + ", dataConverter="
- + dataConverter
+ ", maxConcurrentActivityExecutionSize="
+ maxConcurrentActivityExecutionSize
+ ", maxConcurrentWorkflowExecutionSize="
@@ -313,14 +221,6 @@ public String toString() {
+ activityPollerOptions
+ ", workflowPollerOptions="
+ workflowPollerOptions
- + ", reportActivityCompletionRetryOptions="
- + reportActivityCompletionRetryOptions
- + ", reportActivityFailureRetryOptions="
- + reportActivityFailureRetryOptions
- + ", reportWorkflowCompletionRetryOptions="
- + reportWorkflowCompletionRetryOptions
- + ", reportWorkflowFailureRetryOptions="
- + reportWorkflowFailureRetryOptions
+ '}';
}
}
diff --git a/src/test/java/com/uber/cadence/activity/ActivityOptionsTest.java b/src/test/java/com/uber/cadence/activity/ActivityOptionsTest.java
index bed1e5fc7..654110715 100644
--- a/src/test/java/com/uber/cadence/activity/ActivityOptionsTest.java
+++ b/src/test/java/com/uber/cadence/activity/ActivityOptionsTest.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
@@ -27,65 +27,6 @@
public class ActivityOptionsTest {
- @ActivityMethod
- public void defaultActivityOptions() {}
-
- @Test
- public void testOnlyOptionsPresent() throws NoSuchMethodException {
- ActivityOptions o =
- new ActivityOptions.Builder()
- .setTaskList("foo")
- .setHeartbeatTimeout(Duration.ofSeconds(123))
- .setScheduleToCloseTimeout(Duration.ofSeconds(321))
- .setScheduleToStartTimeout(Duration.ofSeconds(333))
- .setStartToCloseTimeout(Duration.ofSeconds(345))
- .setRetryOptions(
- new RetryOptions.Builder()
- .setDoNotRetry(IllegalArgumentException.class)
- .setMaximumAttempts(11111)
- .setBackoffCoefficient(1.55)
- .setMaximumInterval(Duration.ofDays(3))
- .setExpiration(Duration.ofDays(365))
- .setInitialInterval(Duration.ofMinutes(12))
- .build())
- .build();
- ActivityMethod a =
- ActivityOptionsTest.class
- .getMethod("defaultActivityOptions")
- .getAnnotation(ActivityMethod.class);
- Assert.assertEquals(o, ActivityOptions.merge(a, null, o));
- }
-
- @MethodRetry(initialIntervalSeconds = 3)
- @ActivityMethod
- public void defaultActivityAndRetryOptions() {}
-
- @Test
- public void testOnlyOptionsAndEmptyAnnotationsPresent() throws NoSuchMethodException {
- ActivityOptions o =
- new ActivityOptions.Builder()
- .setTaskList("foo")
- .setHeartbeatTimeout(Duration.ofSeconds(123))
- .setScheduleToCloseTimeout(Duration.ofSeconds(321))
- .setScheduleToStartTimeout(Duration.ofSeconds(333))
- .setStartToCloseTimeout(Duration.ofSeconds(345))
- .setRetryOptions(
- new RetryOptions.Builder()
- .setDoNotRetry(IllegalArgumentException.class)
- .setMaximumAttempts(11111)
- .setBackoffCoefficient(1.55)
- .setMaximumInterval(Duration.ofDays(3))
- .setExpiration(Duration.ofDays(365))
- .setInitialInterval(Duration.ofMinutes(12))
- .build())
- .build();
- ActivityMethod a =
- ActivityOptionsTest.class
- .getMethod("defaultActivityAndRetryOptions")
- .getAnnotation(ActivityMethod.class);
- Assert.assertEquals(o, ActivityOptions.merge(a, null, o));
- }
-
@MethodRetry(
initialIntervalSeconds = 12,
backoffCoefficient = 1.97,
@@ -94,30 +35,14 @@ public void testOnlyOptionsAndEmptyAnnotationsPresent() throws NoSuchMethodExcep
maximumIntervalSeconds = 22,
doNotRetry = {NullPointerException.class, UnsupportedOperationException.class}
)
- @ActivityMethod(
- startToCloseTimeoutSeconds = 1135,
- taskList = "bar",
- heartbeatTimeoutSeconds = 4567,
- scheduleToCloseTimeoutSeconds = 2342,
- scheduleToStartTimeoutSeconds = 9879
- )
public void activityAndRetryOptions() {}
@Test
public void testOnlyAnnotationsPresent() throws NoSuchMethodException {
Method method = ActivityOptionsTest.class.getMethod("activityAndRetryOptions");
- ActivityMethod a = method.getAnnotation(ActivityMethod.class);
MethodRetry r = method.getAnnotation(MethodRetry.class);
- ActivityOptions o = new ActivityOptions.Builder().build();
- ActivityOptions merged = ActivityOptions.merge(a, r, o);
- Assert.assertEquals(a.taskList(), merged.getTaskList());
- Assert.assertEquals(a.heartbeatTimeoutSeconds(), merged.getHeartbeatTimeout().getSeconds());
- Assert.assertEquals(
- a.scheduleToCloseTimeoutSeconds(), merged.getScheduleToCloseTimeout().getSeconds());
- Assert.assertEquals(
- a.scheduleToStartTimeoutSeconds(), merged.getScheduleToStartTimeout().getSeconds());
- Assert.assertEquals(
- a.startToCloseTimeoutSeconds(), merged.getStartToCloseTimeout().getSeconds());
+ ActivityOptions o = ActivityOptions.newBuilder().build();
+ ActivityOptions merged = ActivityOptions.newBuilder(o).setMethodRetry(r).build();
RetryOptions rMerged = merged.getRetryOptions();
Assert.assertEquals(r.maximumAttempts(), rMerged.getMaximumAttempts());
@@ -129,29 +54,4 @@ public void testOnlyAnnotationsPresent() throws NoSuchMethodException {
Duration.ofSeconds(r.maximumIntervalSeconds()), rMerged.getMaximumInterval());
Assert.assertEquals(Arrays.asList(r.doNotRetry()), rMerged.getDoNotRetry());
}
-
- @Test
- public void testBothPresent() throws NoSuchMethodException {
- ActivityOptions o =
- new ActivityOptions.Builder()
- .setTaskList("foo")
- .setHeartbeatTimeout(Duration.ofSeconds(123))
- .setScheduleToCloseTimeout(Duration.ofSeconds(321))
- .setScheduleToStartTimeout(Duration.ofSeconds(333))
- .setStartToCloseTimeout(Duration.ofSeconds(345))
- .setRetryOptions(
- new RetryOptions.Builder()
- .setDoNotRetry(IllegalArgumentException.class)
- .setMaximumAttempts(11111)
- .setBackoffCoefficient(1.55)
- .setMaximumInterval(Duration.ofDays(3))
- .setExpiration(Duration.ofDays(365))
- .setInitialInterval(Duration.ofMinutes(12))
- .build())
- .build();
- Method method = ActivityOptionsTest.class.getMethod("activityAndRetryOptions");
- ActivityMethod a = method.getAnnotation(ActivityMethod.class);
- MethodRetry r = method.getAnnotation(MethodRetry.class);
- Assert.assertEquals(o, ActivityOptions.merge(a, r, o));
- }
}
diff --git a/src/test/java/com/uber/cadence/activity/LocalActivityContextPropagationTest.java b/src/test/java/com/uber/cadence/activity/LocalActivityContextPropagationTest.java
index 6b8817b9e..3eced394c 100644
--- a/src/test/java/com/uber/cadence/activity/LocalActivityContextPropagationTest.java
+++ b/src/test/java/com/uber/cadence/activity/LocalActivityContextPropagationTest.java
@@ -155,6 +155,7 @@ public String foo() {
}
}
+ @ActivityInterface
public interface LocalActivityContextPropagation {
@ActivityMethod
diff --git a/src/test/java/com/uber/cadence/internal/common/RetryerTest.java b/src/test/java/com/uber/cadence/internal/common/RetryerTest.java
index 1cbf543fe..612764e85 100644
--- a/src/test/java/com/uber/cadence/internal/common/RetryerTest.java
+++ b/src/test/java/com/uber/cadence/internal/common/RetryerTest.java
@@ -40,7 +40,7 @@ public void testExpiration() throws InterruptedException {
.validateBuildWithDefaults();
long start = System.currentTimeMillis();
try {
- Retryer.retryWithResultAsync(
+ RpcRetryer.retryWithResultAsync(
options,
() -> {
throw new IllegalArgumentException("simulated");
@@ -64,7 +64,7 @@ public void testExpirationFuture() throws InterruptedException {
.validateBuildWithDefaults();
long start = System.currentTimeMillis();
try {
- Retryer.retryWithResultAsync(
+ RpcRetryer.retryWithResultAsync(
options,
() -> {
CompletableFuture result = new CompletableFuture<>();
@@ -91,7 +91,7 @@ public void testInterruptedException() throws InterruptedException {
.validateBuildWithDefaults();
long start = System.currentTimeMillis();
try {
- Retryer.retryWithResultAsync(
+ RpcRetryer.retryWithResultAsync(
options,
() -> {
CompletableFuture result = new CompletableFuture<>();
@@ -117,7 +117,7 @@ public void testAddDoNotRetry() throws InterruptedException {
options = options.addDoNotRetry(InterruptedException.class);
long start = System.currentTimeMillis();
try {
- Retryer.retryWithResultAsync(
+ RpcRetryer.retryWithResultAsync(
options,
() -> {
CompletableFuture result = new CompletableFuture<>();
@@ -144,7 +144,7 @@ public void testMaxAttempt() throws InterruptedException {
.validateBuildWithDefaults();
long start = System.currentTimeMillis();
try {
- Retryer.retryWithResultAsync(
+ RpcRetryer.retryWithResultAsync(
options,
() -> {
throw new IllegalArgumentException("simulated");
diff --git a/src/test/java/com/uber/cadence/internal/sync/ActivityInterfaceTest.java b/src/test/java/com/uber/cadence/internal/sync/ActivityInterfaceTest.java
new file mode 100644
index 000000000..693f540f9
--- /dev/null
+++ b/src/test/java/com/uber/cadence/internal/sync/ActivityInterfaceTest.java
@@ -0,0 +1,192 @@
+/*
+ * 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.
+ *
+ * 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.google.common.base.Objects;
+import com.uber.cadence.activity.ActivityInterface;
+import java.lang.annotation.Annotation;
+import java.lang.reflect.Method;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Set;
+import org.junit.Test;
+
+public class ActivityInterfaceTest {
+
+ public interface I0 {
+ void baz();
+ }
+
+ public interface I1 extends I0 {
+ void foo();
+ }
+
+ @ActivityInterface
+ public interface I2 extends I1, I0 {
+ void bar();
+ }
+
+ @ActivityInterface
+ public interface I3 extends I1 {
+ void foo();
+ }
+
+ @ActivityInterface
+ public interface I4 extends I3 {
+ void foo();
+
+ void bar();
+ }
+
+ @ActivityInterface
+ public interface I5 extends I4, I3 {}
+
+ public interface NonActivity {
+ void foobar();
+ }
+
+ public static class Impl implements I2, I4, NonActivity, I5 {
+
+ @Override
+ public void foo() {}
+
+ @Override
+ public void bar() {}
+
+ @Override
+ public void foobar() {}
+
+ @Override
+ public void baz() {}
+ }
+
+ @Test
+ public void test() {
+ Set activityMethods =
+ getAnnotatedInterfaceMethods(Impl.class, ActivityInterface.class);
+ for (Iterator iterator = activityMethods.iterator();
+ iterator.hasNext(); ) {
+ MethodInterfacePair next = iterator.next();
+ System.out.println(next);
+ }
+ }
+
+ static class MethodInterfacePair {
+ private final Method method;
+ private final Class type;
+
+ MethodInterfacePair(Method method, Class type) {
+ this.method = method;
+ this.type = type;
+ }
+
+ public Method getMethod() {
+ return method;
+ }
+
+ public Class getType() {
+ return type;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ MethodInterfacePair that = (MethodInterfacePair) o;
+ return Objects.equal(method, that.method) && Objects.equal(type, that.type);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(method, type);
+ }
+
+ @Override
+ public String toString() {
+ return "MethodInterfacePair{" + "method=" + method + ", type=" + type + '}';
+ }
+ }
+
+ /** Used to override equals and hashCode of Method to ensure deduping by method name in a set. */
+ static class MethodWrapper {
+ private final Method method;
+
+ MethodWrapper(Method method) {
+ this.method = method;
+ }
+
+ public Method getMethod() {
+ return method;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ MethodWrapper that = (MethodWrapper) o;
+ return Objects.equal(method.getName(), that.method.getName());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hashCode(method.getName());
+ }
+ }
+
+ Set getAnnotatedInterfaceMethods(
+ Class implementationClass, Class annotationClass) {
+ if (implementationClass.isInterface()) {
+ throw new IllegalArgumentException(
+ "Concrete class expected. Found interface: " + implementationClass.getSimpleName());
+ }
+ Set pairs = new HashSet<>();
+ // Methods inherited from interfaces that are not annotated with @ActivityInterface
+ Set ignored = new HashSet<>();
+ getAnnotatedInterfaceMethods(implementationClass, annotationClass, ignored, pairs);
+ return pairs;
+ }
+
+ private void getAnnotatedInterfaceMethods(
+ Class current,
+ Class annotationClass,
+ Set methods,
+ Set result) {
+ // Using set to dedupe methods which are defined in both non activity parent and current
+ Set ourMethods = new HashSet<>();
+ if (current.isInterface()) {
+ Method[] declaredMethods = current.getDeclaredMethods();
+ for (int i = 0; i < declaredMethods.length; i++) {
+ Method declaredMethod = declaredMethods[i];
+ ourMethods.add(new MethodWrapper(declaredMethod));
+ }
+ }
+ Class[] interfaces = current.getInterfaces();
+ for (int i = 0; i < interfaces.length; i++) {
+ Class anInterface = interfaces[i];
+ getAnnotatedInterfaceMethods(anInterface, annotationClass, ourMethods, result);
+ }
+ Annotation annotation = current.getAnnotation(annotationClass);
+ if (annotation == null) {
+ methods.addAll(ourMethods);
+ return;
+ }
+ for (MethodWrapper method : ourMethods) {
+ result.add(new MethodInterfacePair(method.getMethod(), current));
+ }
+ }
+}
diff --git a/src/test/java/com/uber/cadence/internal/testing/ActivityTestingTest.java b/src/test/java/com/uber/cadence/internal/testing/ActivityTestingTest.java
index 9603f9e4c..f1ff72caa 100644
--- a/src/test/java/com/uber/cadence/internal/testing/ActivityTestingTest.java
+++ b/src/test/java/com/uber/cadence/internal/testing/ActivityTestingTest.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
@@ -26,13 +26,15 @@
import com.uber.cadence.RecordActivityTaskHeartbeatResponse;
import com.uber.cadence.activity.Activity;
-import com.uber.cadence.activity.ActivityMethod;
+import com.uber.cadence.activity.ActivityInterface;
import com.uber.cadence.client.ActivityCancelledException;
import com.uber.cadence.serviceclient.IWorkflowService;
import com.uber.cadence.testing.TestActivityEnvironment;
import com.uber.cadence.workflow.ActivityFailureException;
import io.netty.util.internal.ConcurrentSet;
import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.thrift.TException;
@@ -48,6 +50,7 @@ public void setUp() {
testEnvironment = TestActivityEnvironment.newInstance();
}
+ @ActivityInterface
public interface TestActivity {
String activity1(String input);
@@ -111,9 +114,8 @@ public void testHeartbeat() {
assertEquals("details1", details.get());
}
+ @ActivityInterface
public interface InterruptibleTestActivity {
-
- @ActivityMethod(scheduleToCloseTimeoutSeconds = 1000, heartbeatTimeoutSeconds = 1)
void activity1() throws InterruptedException;
}
@@ -252,4 +254,63 @@ public void testHeartbeatIntermittentError() throws TException, InterruptedExcep
activity.activity1();
assertEquals(3, count.get());
}
+
+ public interface A {
+ void a();
+ }
+
+ @ActivityInterface
+ public interface B extends A {
+ void b();
+ }
+
+ @ActivityInterface
+ public interface C extends B, A {
+ void c();
+ }
+
+ public class CImpl implements C {
+ private List invocations = new ArrayList<>();
+
+ @Override
+ public void a() {
+ invocations.add("a");
+ }
+
+ @Override
+ public void b() {
+ invocations.add("b");
+ }
+
+ @Override
+ public void c() {
+ invocations.add("c");
+ }
+ }
+
+ @Test
+ public void testInvokingActivityByBaseInterface() {
+ CImpl impl = new CImpl();
+ testEnvironment.registerActivitiesImplementations(impl);
+ try {
+ testEnvironment.newActivityStub(A.class);
+ fail("A doesn't implement activity");
+ } catch (IllegalArgumentException e) {
+ // expected as A doesn't implement any activity
+ }
+ B b = testEnvironment.newActivityStub(B.class);
+ b.a();
+ b.b();
+ C c = testEnvironment.newActivityStub(C.class);
+ c.a();
+ c.b();
+ c.c();
+ List expected = new ArrayList<>();
+ expected.add("a");
+ expected.add("b");
+ expected.add("a");
+ expected.add("b");
+ expected.add("c");
+ assertEquals(expected, impl.invocations);
+ }
}
diff --git a/src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java b/src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java
index e22de1bdd..27d8ac4eb 100644
--- a/src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.java
+++ b/src/test/java/com/uber/cadence/internal/testing/WorkflowTestingTest.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
@@ -36,7 +36,7 @@
import com.uber.cadence.WorkflowExecution;
import com.uber.cadence.WorkflowExecutionInfo;
import com.uber.cadence.activity.Activity;
-import com.uber.cadence.activity.ActivityMethod;
+import com.uber.cadence.activity.ActivityInterface;
import com.uber.cadence.activity.ActivityOptions;
import com.uber.cadence.client.*;
import com.uber.cadence.context.ContextPropagator;
@@ -162,9 +162,8 @@ public void testFailure() {
}
}
+ @ActivityInterface
public interface TestActivity {
-
- @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
String activity1(String input);
}
@@ -178,7 +177,10 @@ public String activity1(String input) {
public static class ActivityWorkflow implements TestWorkflow {
- private final TestActivity activity = Workflow.newActivityStub(TestActivity.class);
+ private final TestActivity activity =
+ Workflow.newActivityStub(
+ TestActivity.class,
+ ActivityOptions.newBuilder().setScheduleToCloseTimeout(Duration.ofHours(1)).build());
@Override
public String workflow1(String input) {
@@ -486,9 +488,8 @@ public void testConcurrentDecision() throws ExecutionException, InterruptedExcep
log.info(testEnvironment.getDiagnostics());
}
+ @ActivityInterface
public interface TestCancellationActivity {
-
- @ActivityMethod(scheduleToCloseTimeoutSeconds = 1000, heartbeatTimeoutSeconds = 2)
String activity1(String input);
}
@@ -506,7 +507,12 @@ public String activity1(String input) {
public static class TestCancellationWorkflow implements TestWorkflow {
private final TestCancellationActivity activity =
- Workflow.newActivityStub(TestCancellationActivity.class);
+ Workflow.newActivityStub(
+ TestCancellationActivity.class,
+ ActivityOptions.newBuilder()
+ .setScheduleToCloseTimeout(Duration.ofSeconds(1000))
+ .setHeartbeatTimeout(Duration.ofSeconds(2))
+ .build());
@Override
public String workflow1(String input) {
@@ -542,7 +548,11 @@ public static class TestTimerCancellationWorkflow implements TestWorkflow {
public String workflow1(String input) {
long startTime = Workflow.currentTimeMillis();
Promise s = Async.procedure(() -> Workflow.sleep(Duration.ofHours(3)));
- TestActivity activity = Workflow.newActivityStub(TestActivity.class);
+ TestActivity activity =
+ Workflow.newActivityStub(
+ TestActivity.class,
+ ActivityOptions.newBuilder().setScheduleToCloseTimeout(Duration.ofHours(1)).build());
+
activity.activity1("input");
Workflow.sleep(Duration.ofHours(1));
s.get();
@@ -894,7 +904,11 @@ public String workflow1(String input) {
.setScheduleToCloseTimeout(Duration.ofSeconds(5))
.setContextPropagators(Collections.singletonList(new TestContextPropagator()))
.build();
- TestActivity activity = Workflow.newActivityStub(TestActivity.class, options);
+ TestActivity activity =
+ Workflow.newActivityStub(
+ TestActivity.class,
+ ActivityOptions.newBuilder().setScheduleToCloseTimeout(Duration.ofHours(1)).build());
+
return activity.activity1("foo");
}
}
diff --git a/src/test/java/com/uber/cadence/worker/CleanWorkerShutdownTest.java b/src/test/java/com/uber/cadence/worker/CleanWorkerShutdownTest.java
index 936370cd1..6235b53e9 100644
--- a/src/test/java/com/uber/cadence/worker/CleanWorkerShutdownTest.java
+++ b/src/test/java/com/uber/cadence/worker/CleanWorkerShutdownTest.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
@@ -27,7 +27,8 @@
import com.uber.cadence.HistoryEvent;
import com.uber.cadence.WorkflowExecution;
import com.uber.cadence.activity.Activity;
-import com.uber.cadence.activity.ActivityMethod;
+import com.uber.cadence.activity.ActivityInterface;
+import com.uber.cadence.activity.ActivityOptions;
import com.uber.cadence.client.ActivityWorkerShutdownException;
import com.uber.cadence.client.WorkflowClient;
import com.uber.cadence.client.WorkflowClientOptions;
@@ -40,6 +41,7 @@
import com.uber.cadence.workflow.Workflow;
import com.uber.cadence.workflow.WorkflowMethod;
import java.nio.charset.StandardCharsets;
+import java.time.Duration;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -97,7 +99,12 @@ public interface TestWorkflow {
public static class TestWorkflowImpl implements TestWorkflow {
- private final Activities activities = Workflow.newActivityStub(Activities.class);
+ private final Activities activities =
+ Workflow.newActivityStub(
+ Activities.class,
+ ActivityOptions.newBuilder()
+ .setScheduleToCloseTimeout(Duration.ofSeconds(100))
+ .build());
@Override
public String execute() {
@@ -105,8 +112,8 @@ public String execute() {
}
}
+ @ActivityInterface
public interface Activities {
- @ActivityMethod(scheduleToCloseTimeoutSeconds = 100)
String execute();
}
diff --git a/src/test/java/com/uber/cadence/worker/StickyWorkerTest.java b/src/test/java/com/uber/cadence/worker/StickyWorkerTest.java
index 8a2ecbf5e..b07b7ef1a 100644
--- a/src/test/java/com/uber/cadence/worker/StickyWorkerTest.java
+++ b/src/test/java/com/uber/cadence/worker/StickyWorkerTest.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
@@ -28,7 +28,7 @@
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
-import com.uber.cadence.activity.ActivityMethod;
+import com.uber.cadence.activity.ActivityInterface;
import com.uber.cadence.activity.ActivityOptions;
import com.uber.cadence.client.WorkflowClient;
import com.uber.cadence.client.WorkflowClientOptions;
@@ -709,9 +709,8 @@ public void execute(WorkflowParams params) {
}
}
+ @ActivityInterface
public interface SleepActivity {
-
- @ActivityMethod()
void sleep(int chain, int concurrency, byte[] bytes);
}
diff --git a/src/test/java/com/uber/cadence/worker/WorkerStressTests.java b/src/test/java/com/uber/cadence/worker/WorkerStressTests.java
index 30508fbda..7ecdf5fad 100644
--- a/src/test/java/com/uber/cadence/worker/WorkerStressTests.java
+++ b/src/test/java/com/uber/cadence/worker/WorkerStressTests.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
@@ -20,7 +20,7 @@
import static com.uber.cadence.workflow.WorkflowTest.DOMAIN;
import static org.junit.Assert.assertNotNull;
-import com.uber.cadence.activity.ActivityMethod;
+import com.uber.cadence.activity.ActivityInterface;
import com.uber.cadence.activity.ActivityOptions;
import com.uber.cadence.client.WorkflowClient;
import com.uber.cadence.client.WorkflowClientOptions;
@@ -260,9 +260,8 @@ public String execute(WorkflowParams params) {
}
}
+ @ActivityInterface
public interface SleepActivity {
-
- @ActivityMethod()
void sleep(int chain, int concurrency, byte[] bytes);
}
diff --git a/src/test/java/com/uber/cadence/workflow/MetricsTest.java b/src/test/java/com/uber/cadence/workflow/MetricsTest.java
index 21417847a..9010d2dfa 100644
--- a/src/test/java/com/uber/cadence/workflow/MetricsTest.java
+++ b/src/test/java/com/uber/cadence/workflow/MetricsTest.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
@@ -22,6 +22,7 @@
import static org.mockito.Matchers.eq;
import static org.mockito.Mockito.*;
+import com.uber.cadence.activity.ActivityInterface;
import com.uber.cadence.activity.ActivityOptions;
import com.uber.cadence.client.WorkflowClient;
import com.uber.cadence.client.WorkflowClientOptions;
@@ -101,6 +102,7 @@ public void execute() {
}
}
+ @ActivityInterface
public interface TestActivity {
int runActivity(int input);
}
diff --git a/src/test/java/com/uber/cadence/workflow/TestEnvironmentCloseTest.java b/src/test/java/com/uber/cadence/workflow/TestEnvironmentCloseTest.java
index 7c5692554..54d5fba21 100644
--- a/src/test/java/com/uber/cadence/workflow/TestEnvironmentCloseTest.java
+++ b/src/test/java/com/uber/cadence/workflow/TestEnvironmentCloseTest.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,7 @@
import static junit.framework.TestCase.assertTrue;
-import com.uber.cadence.activity.ActivityMethod;
+import com.uber.cadence.activity.ActivityInterface;
import com.uber.cadence.testing.TestWorkflowEnvironment;
import com.uber.cadence.worker.Worker;
import org.junit.Test;
@@ -43,8 +43,8 @@ public void foo() {}
public void signal() {}
}
+ @ActivityInterface
public interface A {
- @ActivityMethod
void bar();
}
diff --git a/src/test/java/com/uber/cadence/workflow/WorkflowTest.java b/src/test/java/com/uber/cadence/workflow/WorkflowTest.java
index aa24f0cd7..63bc010e4 100644
--- a/src/test/java/com/uber/cadence/workflow/WorkflowTest.java
+++ b/src/test/java/com/uber/cadence/workflow/WorkflowTest.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
@@ -41,11 +41,7 @@
import com.uber.cadence.WorkflowExecutionAlreadyStartedError;
import com.uber.cadence.WorkflowExecutionCloseStatus;
import com.uber.cadence.WorkflowIdReusePolicy;
-import com.uber.cadence.activity.Activity;
-import com.uber.cadence.activity.ActivityMethod;
-import com.uber.cadence.activity.ActivityOptions;
-import com.uber.cadence.activity.ActivityTask;
-import com.uber.cadence.activity.LocalActivityOptions;
+import com.uber.cadence.activity.*;
import com.uber.cadence.client.ActivityCancelledException;
import com.uber.cadence.client.ActivityCompletionClient;
import com.uber.cadence.client.ActivityNotExistsException;
@@ -73,7 +69,11 @@
import com.uber.cadence.testing.TestEnvironmentOptions;
import com.uber.cadence.testing.TestWorkflowEnvironment;
import com.uber.cadence.testing.WorkflowReplayer;
-import com.uber.cadence.worker.*;
+import com.uber.cadence.worker.Worker;
+import com.uber.cadence.worker.WorkerFactory;
+import com.uber.cadence.worker.WorkerFactoryOptions;
+import com.uber.cadence.worker.WorkerOptions;
+import com.uber.cadence.worker.WorkflowImplementationOptions;
import com.uber.cadence.workflow.Functions.Func;
import com.uber.cadence.workflow.Functions.Func1;
import java.io.File;
@@ -771,7 +771,15 @@ public static class TestActivityRetryAnnotated implements TestWorkflow1 {
private final TestActivities activities;
public TestActivityRetryAnnotated() {
- this.activities = Workflow.newActivityStub(TestActivities.class);
+ this.activities =
+ Workflow.newActivityStub(
+ TestActivities.class,
+ ActivityOptions.newBuilder()
+ .setScheduleToCloseTimeout(Duration.ofSeconds(5))
+ .setScheduleToStartTimeout(Duration.ofSeconds(5))
+ .setHeartbeatTimeout(Duration.ofSeconds(5))
+ .setStartToCloseTimeout(Duration.ofSeconds(10))
+ .build());
}
@Override
@@ -2933,9 +2941,10 @@ public String execute(String taskList) {
}
}
+ @ActivityInterface
public interface AngryChildActivity {
- @ActivityMethod(scheduleToCloseTimeoutSeconds = 5)
+ @ActivityMethod
void execute();
}
@@ -2960,7 +2969,10 @@ public String execute(String taskList, int delay) {
AngryChildActivity activity =
Workflow.newActivityStub(
AngryChildActivity.class,
- new ActivityOptions.Builder().setTaskList(taskList).build());
+ ActivityOptions.newBuilder()
+ .setTaskList(taskList)
+ .setScheduleToCloseTimeout(Duration.ofSeconds(5))
+ .build());
activity.execute();
throw new UnsupportedOperationException("simulated failure");
}
@@ -3663,6 +3675,7 @@ public void testChildWorkflowWithCronSchedule() {
Assert.assertEquals("run 2", lastCompletionResult);
}
+ @ActivityInterface
public interface TestActivities {
String sleepActivity(long milliseconds, int input);
@@ -3704,12 +3717,6 @@ public interface TestActivities {
void neverComplete();
- @ActivityMethod(
- scheduleToStartTimeoutSeconds = 5,
- scheduleToCloseTimeoutSeconds = 5,
- heartbeatTimeoutSeconds = 5,
- startToCloseTimeoutSeconds = 10
- )
@MethodRetry(
initialIntervalSeconds = 1,
maximumIntervalSeconds = 1,
@@ -4750,6 +4757,7 @@ public void testUUIDAndRandom() {
tracer.setExpected("sideEffect", "sideEffect", "executeActivity TestActivities::activity2");
}
+ @ActivityInterface
public interface GenericParametersActivity {
List execute(List arg1, Set arg2);
@@ -4858,9 +4866,8 @@ public NonSerializableException() {
}
}
+ @ActivityInterface
public interface NonSerializableExceptionActivity {
-
- @ActivityMethod(scheduleToCloseTimeoutSeconds = 5)
void execute();
}
@@ -4878,7 +4885,11 @@ public static class TestNonSerializableExceptionInActivityWorkflow implements Te
@Override
public String execute(String taskList) {
NonSerializableExceptionActivity activity =
- Workflow.newActivityStub(NonSerializableExceptionActivity.class);
+ Workflow.newActivityStub(
+ NonSerializableExceptionActivity.class,
+ ActivityOptions.newBuilder()
+ .setScheduleToCloseTimeout(Duration.ofSeconds(5))
+ .build());
try {
activity.execute();
} catch (ActivityFailureException e) {
@@ -4900,9 +4911,8 @@ public void testNonSerializableExceptionInActivity() {
assertTrue(result.contains("NonSerializableException"));
}
+ @ActivityInterface
public interface NonDeserializableArgumentsActivity {
-
- @ActivityMethod(scheduleToCloseTimeoutSeconds = 5)
void execute(int arg);
}
@@ -5431,8 +5441,9 @@ public interface GreetingWorkflow {
void createGreeting(String name);
}
+ @ActivityInterface
public interface GreetingActivities {
- @ActivityMethod(scheduleToCloseTimeoutSeconds = 60)
+ @ActivityMethod
String composeGreeting(String string);
}
@@ -5452,7 +5463,9 @@ public String composeGreeting(String string) {
public static class TimerFiringWorkflowImpl implements GreetingWorkflow {
private final GreetingActivities activities =
- Workflow.newActivityStub(GreetingActivities.class);
+ Workflow.newActivityStub(
+ GreetingActivities.class,
+ ActivityOptions.newBuilder().setScheduleToCloseTimeout(Duration.ofSeconds(5)).build());
@Override
public void createGreeting(String name) {