Skip to content

Commit 1a2c896

Browse files
abhishekj720abhishekj720
authored
Merge pull request #84 from uber/nond2
Added replayer test for continue as new
2 parents 2bce1c5 + 3fd8caf commit 1a2c896

File tree

4 files changed

+358
-8
lines changed

4 files changed

+358
-8
lines changed

‎src/test/java/com/uber/cadence/samples/replaytests/HelloActivityReplayTest.java

-8
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,6 @@
1818
package com.uber.cadence.samples.replaytests;
1919

2020
import com.uber.cadence.samples.hello.HelloActivity;
21-
import com.uber.cadence.samples.hello.HelloPeriodic;
2221
import com.uber.cadence.testing.WorkflowReplayer;
2322
import org.junit.Test;
2423

@@ -38,11 +37,4 @@ public void testReplay() throws Exception {
3837
WorkflowReplayer.replayWorkflowExecutionFromResource(
3938
"replaytests/HelloActivity.json", HelloActivity.GreetingWorkflowImpl.class);
4039
}
41-
42-
// continue-as-new case for replayer tests
43-
@Test
44-
public void testReplay_continueAsNew() throws Exception {
45-
WorkflowReplayer.replayWorkflowExecutionFromResource(
46-
"replaytests/HelloPeriodic.json", HelloPeriodic.GreetingWorkflowImpl.class);
47-
}
4840
}

‎src/test/java/com/uber/cadence/samples/replaytests/HelloPeriodicReplayTest.java

+49
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
/*
2+
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Modifications copyright (C) 2017 Uber Technologies, Inc.
5+
*
6+
* Licensed under the Apache License, Version 2.0 (the "License"). You may not
7+
* use this file except in compliance with the License. A copy of the License is
8+
* located at
9+
*
10+
* http://aws.amazon.com/apache2.0
11+
*
12+
* or in the "license" file accompanying this file. This file is distributed on
13+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14+
* express or implied. See the License for the specific language governing
15+
* permissions and limitations under the License.
16+
*/
17+
18+
package com.uber.cadence.samples.replaytests;
19+
20+
import com.uber.cadence.samples.hello.HelloPeriodic;
21+
import com.uber.cadence.testing.WorkflowReplayer;
22+
import org.junit.Test;
23+
24+
public class HelloPeriodicReplayTest {
25+
26+
/* Runs a history which ends with WorkflowExecutionContinuedAsNew. Replay fails because of the additional checks done for continue as new case by replayWorkflowHistory(). This should not have any error because it's a valid continue as new case. */
27+
@Test
28+
public void testReplay_continueAsNew() throws Exception {
29+
WorkflowReplayer.replayWorkflowExecutionFromResource(
30+
"replaytests/HelloPeriodic.json", HelloPeriodic.GreetingWorkflowImpl.class);
31+
}
32+
33+
// Continue as new case: change in frequency compared to original workflow definition by
34+
// increasing number of times greet is hit. It should
35+
// fail. BUT it is currently passing.
36+
@Test
37+
public void testReplay_continueAsNew_moreFrequency() throws Exception {
38+
WorkflowReplayer.replayWorkflowExecutionFromResource(
39+
"replaytests/HelloPeriodic.json", HelloPeriodic_moreFrequency.GreetingWorkflowImpl.class);
40+
}
41+
42+
// Continue as new case: If frequency is changed to lesser number.
43+
// FAIL As expected: It should hit non-determinism case and it is hitting properly.
44+
// @Test
45+
// public void testReplay_continueAsNew_lessFrequency() throws Exception {
46+
// WorkflowReplayer.replayWorkflowExecutionFromResource(
47+
// "replaytests/HelloPeriodic.json", HelloPeriodic_lessFrequency.GreetingWorkflowImpl.class);
48+
// }
49+
}

‎src/test/java/com/uber/cadence/samples/replaytests/HelloPeriodic_lessFrequency.java

+155
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,155 @@
1+
/*
2+
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Modifications copyright (C) 2017 Uber Technologies, Inc.
5+
*
6+
* Licensed under the Apache License, Version 2.0 (the "License"). You may not
7+
* use this file except in compliance with the License. A copy of the License is
8+
* located at
9+
*
10+
* http://aws.amazon.com/apache2.0
11+
*
12+
* or in the "license" file accompanying this file. This file is distributed on
13+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14+
* express or implied. See the License for the specific language governing
15+
* permissions and limitations under the License.
16+
*/
17+
18+
package com.uber.cadence.samples.replaytests;
19+
20+
import static com.uber.cadence.samples.common.SampleConstants.DOMAIN;
21+
22+
import com.google.common.base.Throwables;
23+
import com.uber.cadence.WorkflowExecution;
24+
import com.uber.cadence.WorkflowIdReusePolicy;
25+
import com.uber.cadence.activity.Activity;
26+
import com.uber.cadence.activity.ActivityOptions;
27+
import com.uber.cadence.client.DuplicateWorkflowException;
28+
import com.uber.cadence.client.WorkflowClient;
29+
import com.uber.cadence.client.WorkflowClientOptions;
30+
import com.uber.cadence.client.WorkflowException;
31+
import com.uber.cadence.client.WorkflowStub;
32+
import com.uber.cadence.internal.compatibility.Thrift2ProtoAdapter;
33+
import com.uber.cadence.internal.compatibility.proto.serviceclient.IGrpcServiceStubs;
34+
import com.uber.cadence.worker.Worker;
35+
import com.uber.cadence.worker.WorkerFactory;
36+
import com.uber.cadence.workflow.Workflow;
37+
import com.uber.cadence.workflow.WorkflowMethod;
38+
import java.time.Duration;
39+
import java.util.Optional;
40+
41+
public class HelloPeriodic_lessFrequency {
42+
43+
static final String TASK_LIST = "HelloPeriodic";
44+
static final String PERIODIC_WORKFLOW_ID = "HelloPeriodic";
45+
46+
public interface GreetingWorkflow {
47+
@WorkflowMethod(
48+
// At most one instance.
49+
workflowId = PERIODIC_WORKFLOW_ID,
50+
// To allow starting workflow with the same ID after the previous one has terminated.
51+
workflowIdReusePolicy = WorkflowIdReusePolicy.AllowDuplicate,
52+
// Adjust this value to the maximum time workflow is expected to run.
53+
// It usually depends on the number of repetitions and interval between them.
54+
executionStartToCloseTimeoutSeconds = 300,
55+
taskList = TASK_LIST
56+
)
57+
void greetPeriodically(String name, Duration delay);
58+
}
59+
60+
public interface GreetingActivities {
61+
void greet(String greeting);
62+
}
63+
64+
public static class GreetingWorkflowImpl implements GreetingWorkflow {
65+
66+
// If we change the value to 10 (compared to 1000 in original case), then non-determinism case
67+
// is not hitting.
68+
private final int CONTINUE_AS_NEW_FREQUENCEY = 1;
69+
70+
private final GreetingActivities activities =
71+
Workflow.newActivityStub(
72+
GreetingActivities.class,
73+
new ActivityOptions.Builder()
74+
.setScheduleToCloseTimeout(Duration.ofSeconds(10))
75+
.build());
76+
77+
/**
78+
* Stub used to terminate this workflow run and create the next one with the same ID atomically.
79+
*/
80+
private final GreetingWorkflow continueAsNew =
81+
Workflow.newContinueAsNewStub(GreetingWorkflow.class);
82+
83+
@Override
84+
public void greetPeriodically(String name, Duration delay) {
85+
// Loop the predefined number of times then continue this workflow as new.
86+
// This is needed to periodically truncate the history size.
87+
for (int i = 0; i < CONTINUE_AS_NEW_FREQUENCEY; i++) {
88+
activities.greet("Hello " + name + "!");
89+
Workflow.sleep(delay);
90+
}
91+
// Current workflow run stops executing after this call.
92+
continueAsNew.greetPeriodically(name, delay);
93+
// unreachable line
94+
}
95+
}
96+
97+
static class GreetingActivitiesImpl implements GreetingActivities {
98+
@Override
99+
public void greet(String greeting) {
100+
System.out.println("From " + Activity.getWorkflowExecution() + ": " + greeting);
101+
}
102+
}
103+
104+
public static void main(String[] args) throws InterruptedException {
105+
// Get a new client
106+
// NOTE: to set a different options, you can do like this:
107+
// ClientOptions.newBuilder().setRpcTimeout(5 * 1000).build();
108+
WorkflowClient workflowClient =
109+
WorkflowClient.newInstance(
110+
new Thrift2ProtoAdapter(IGrpcServiceStubs.newInstance()),
111+
WorkflowClientOptions.newBuilder().setDomain(DOMAIN).build());
112+
// Get worker to poll the task list.
113+
WorkerFactory factory = WorkerFactory.newInstance(workflowClient);
114+
Worker worker = factory.newWorker(TASK_LIST);
115+
// Workflows are stateful. So you need a type to create instances.
116+
worker.registerWorkflowImplementationTypes(GreetingWorkflowImpl.class);
117+
// Activities are stateless and thread safe. So a shared instance is used.
118+
worker.registerActivitiesImplementations(new GreetingActivitiesImpl());
119+
// Start listening to the workflow and activity task lists.
120+
factory.start();
121+
122+
// Start a workflow execution. Usually this is done from another program.
123+
// To ensure that this daemon type workflow is always running try to start it periodically
124+
// ignoring the duplicated exception.
125+
// It is only to protect from application level failures.
126+
// Failures of a workflow worker don't lead to workflow failures.
127+
WorkflowExecution execution = null;
128+
while (true) {
129+
// Print reason of failure of the previous run, before restarting.
130+
if (execution != null) {
131+
WorkflowStub workflow = workflowClient.newUntypedWorkflowStub(execution, Optional.empty());
132+
try {
133+
workflow.getResult(Void.class); //
134+
} catch (WorkflowException e) {
135+
System.out.println("Previous instance failed:\n" + Throwables.getStackTraceAsString(e));
136+
}
137+
}
138+
// New stub instance should be created for each new workflow start.
139+
GreetingWorkflow workflow = workflowClient.newWorkflowStub(GreetingWorkflow.class);
140+
try {
141+
execution =
142+
WorkflowClient.start(workflow::greetPeriodically, "World", Duration.ofSeconds(3));
143+
System.out.println("Started " + execution);
144+
} catch (DuplicateWorkflowException e) {
145+
System.out.println("Still running as " + e.getExecution());
146+
} catch (Throwable e) {
147+
e.printStackTrace();
148+
System.exit(1);
149+
}
150+
// This value is so low just for the sample purpose. In production workflow
151+
// it is usually much higher.
152+
Thread.sleep(10000);
153+
}
154+
}
155+
}

‎src/test/java/com/uber/cadence/samples/replaytests/HelloPeriodic_moreFrequency.java

+154
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
/*
2+
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved.
3+
*
4+
* Modifications copyright (C) 2017 Uber Technologies, Inc.
5+
*
6+
* Licensed under the Apache License, Version 2.0 (the "License"). You may not
7+
* use this file except in compliance with the License. A copy of the License is
8+
* located at
9+
*
10+
* http://aws.amazon.com/apache2.0
11+
*
12+
* or in the "license" file accompanying this file. This file is distributed on
13+
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
14+
* express or implied. See the License for the specific language governing
15+
* permissions and limitations under the License.
16+
*/
17+
18+
package com.uber.cadence.samples.replaytests;
19+
20+
import static com.uber.cadence.samples.common.SampleConstants.DOMAIN;
21+
22+
import com.google.common.base.Throwables;
23+
import com.uber.cadence.WorkflowExecution;
24+
import com.uber.cadence.WorkflowIdReusePolicy;
25+
import com.uber.cadence.activity.Activity;
26+
import com.uber.cadence.activity.ActivityOptions;
27+
import com.uber.cadence.client.DuplicateWorkflowException;
28+
import com.uber.cadence.client.WorkflowClient;
29+
import com.uber.cadence.client.WorkflowClientOptions;
30+
import com.uber.cadence.client.WorkflowException;
31+
import com.uber.cadence.client.WorkflowStub;
32+
import com.uber.cadence.internal.compatibility.Thrift2ProtoAdapter;
33+
import com.uber.cadence.internal.compatibility.proto.serviceclient.IGrpcServiceStubs;
34+
import com.uber.cadence.worker.Worker;
35+
import com.uber.cadence.worker.WorkerFactory;
36+
import com.uber.cadence.workflow.Workflow;
37+
import com.uber.cadence.workflow.WorkflowMethod;
38+
import java.time.Duration;
39+
import java.util.Optional;
40+
41+
public class HelloPeriodic_moreFrequency {
42+
43+
static final String TASK_LIST = "HelloPeriodic";
44+
static final String PERIODIC_WORKFLOW_ID = "HelloPeriodic";
45+
46+
public interface GreetingWorkflow {
47+
@WorkflowMethod(
48+
// At most one instance.
49+
workflowId = PERIODIC_WORKFLOW_ID,
50+
// To allow starting workflow with the same ID after the previous one has terminated.
51+
workflowIdReusePolicy = WorkflowIdReusePolicy.AllowDuplicate,
52+
// Adjust this value to the maximum time workflow is expected to run.
53+
// It usually depends on the number of repetitions and interval between them.
54+
executionStartToCloseTimeoutSeconds = 300,
55+
taskList = TASK_LIST
56+
)
57+
void greetPeriodically(String name, Duration delay);
58+
}
59+
60+
public interface GreetingActivities {
61+
void greet(String greeting);
62+
}
63+
64+
public static class GreetingWorkflowImpl implements GreetingWorkflow {
65+
66+
// If we change the value to 1, then non-determinism case will hit.
67+
private final int CONTINUE_AS_NEW_FREQUENCEY = 1000;
68+
69+
private final GreetingActivities activities =
70+
Workflow.newActivityStub(
71+
GreetingActivities.class,
72+
new ActivityOptions.Builder()
73+
.setScheduleToCloseTimeout(Duration.ofSeconds(10))
74+
.build());
75+
76+
/**
77+
* Stub used to terminate this workflow run and create the next one with the same ID atomically.
78+
*/
79+
private final GreetingWorkflow continueAsNew =
80+
Workflow.newContinueAsNewStub(GreetingWorkflow.class);
81+
82+
@Override
83+
public void greetPeriodically(String name, Duration delay) {
84+
// Loop the predefined number of times then continue this workflow as new.
85+
// This is needed to periodically truncate the history size.
86+
for (int i = 0; i < CONTINUE_AS_NEW_FREQUENCEY; i++) {
87+
activities.greet("Hello " + name + "!");
88+
Workflow.sleep(delay);
89+
}
90+
// Current workflow run stops executing after this call.
91+
continueAsNew.greetPeriodically(name, delay);
92+
// unreachable line
93+
}
94+
}
95+
96+
static class GreetingActivitiesImpl implements GreetingActivities {
97+
@Override
98+
public void greet(String greeting) {
99+
System.out.println("From " + Activity.getWorkflowExecution() + ": " + greeting);
100+
}
101+
}
102+
103+
public static void main(String[] args) throws InterruptedException {
104+
// Get a new client
105+
// NOTE: to set a different options, you can do like this:
106+
// ClientOptions.newBuilder().setRpcTimeout(5 * 1000).build();
107+
WorkflowClient workflowClient =
108+
WorkflowClient.newInstance(
109+
new Thrift2ProtoAdapter(IGrpcServiceStubs.newInstance()),
110+
WorkflowClientOptions.newBuilder().setDomain(DOMAIN).build());
111+
// Get worker to poll the task list.
112+
WorkerFactory factory = WorkerFactory.newInstance(workflowClient);
113+
Worker worker = factory.newWorker(TASK_LIST);
114+
// Workflows are stateful. So you need a type to create instances.
115+
worker.registerWorkflowImplementationTypes(GreetingWorkflowImpl.class);
116+
// Activities are stateless and thread safe. So a shared instance is used.
117+
worker.registerActivitiesImplementations(new GreetingActivitiesImpl());
118+
// Start listening to the workflow and activity task lists.
119+
factory.start();
120+
121+
// Start a workflow execution. Usually this is done from another program.
122+
// To ensure that this daemon type workflow is always running try to start it periodically
123+
// ignoring the duplicated exception.
124+
// It is only to protect from application level failures.
125+
// Failures of a workflow worker don't lead to workflow failures.
126+
WorkflowExecution execution = null;
127+
while (true) {
128+
// Print reason of failure of the previous run, before restarting.
129+
if (execution != null) {
130+
WorkflowStub workflow = workflowClient.newUntypedWorkflowStub(execution, Optional.empty());
131+
try {
132+
workflow.getResult(Void.class); //
133+
} catch (WorkflowException e) {
134+
System.out.println("Previous instance failed:\n" + Throwables.getStackTraceAsString(e));
135+
}
136+
}
137+
// New stub instance should be created for each new workflow start.
138+
GreetingWorkflow workflow = workflowClient.newWorkflowStub(GreetingWorkflow.class);
139+
try {
140+
execution =
141+
WorkflowClient.start(workflow::greetPeriodically, "World", Duration.ofSeconds(3));
142+
System.out.println("Started " + execution);
143+
} catch (DuplicateWorkflowException e) {
144+
System.out.println("Still running as " + e.getExecution());
145+
} catch (Throwable e) {
146+
e.printStackTrace();
147+
System.exit(1);
148+
}
149+
// This value is so low just for the sample purpose. In production workflow
150+
// it is usually much higher.
151+
Thread.sleep(10000);
152+
}
153+
}
154+
}

0 commit comments

Comments
 (0)