Skip to content

Commit 626bf03

Browse files
SushisourceSushisource
authored
Create worker deployment based versioning sample (#754)
1 parent 0dbfe22 commit 626bf03

15 files changed

+617
-2
lines changed

‎.gitignore‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17,4 +17,5 @@ target
1717
.project
1818
.settings/
1919
bin/
20-
core/.vscode/
20+
core/.vscode/
21+
.claude/

‎README.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,8 +108,9 @@ See the README.md file in each main sample directory for cut/paste Gradle comman
108108

109109
- [**Custom Annotation**](/core/src/main/java/io/temporal/samples/customannotation): Demonstrates how to create a custom annotation using an interceptor.
110110

111-
- [**Asnyc Packet Delivery**](/core/src/main/java/io/temporal/samples/packetdelivery): Demonstrates running multiple execution paths async within single execution.
111+
- [**Async Packet Delivery**](/core/src/main/java/io/temporal/samples/packetdelivery): Demonstrates running multiple execution paths async within single execution.
112112

113+
- [**Worker Versioning**](/core/src/main/java/io/temporal/samples/workerversioning): Demonstrates how to use worker versioning to manage workflow code changes.
113114

114115
#### API demonstrations
115116

‎core/src/main/java/io/temporal/samples/workerversioning/Activities.java‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,38 @@
1+
package io.temporal.samples.workerversioning;
2+
3+
import com.fasterxml.jackson.annotation.JsonCreator;
4+
import com.fasterxml.jackson.annotation.JsonProperty;
5+
import io.temporal.activity.ActivityInterface;
6+
import io.temporal.activity.ActivityMethod;
7+
8+
@ActivityInterface
9+
public interface Activities {
10+
11+
@ActivityMethod
12+
String someActivity(String calledBy);
13+
14+
@ActivityMethod
15+
String someIncompatibleActivity(IncompatibleActivityInput input);
16+
17+
class IncompatibleActivityInput {
18+
private final String calledBy;
19+
private final String moreData;
20+
21+
@JsonCreator(mode = JsonCreator.Mode.PROPERTIES)
22+
public IncompatibleActivityInput(
23+
@JsonProperty("calledBy") String calledBy, @JsonProperty("moreData") String moreData) {
24+
this.calledBy = calledBy;
25+
this.moreData = moreData;
26+
}
27+
28+
@JsonProperty("calledBy")
29+
public String getCalledBy() {
30+
return calledBy;
31+
}
32+
33+
@JsonProperty("moreData")
34+
public String getMoreData() {
35+
return moreData;
36+
}
37+
}
38+
}

‎core/src/main/java/io/temporal/samples/workerversioning/ActivitiesImpl.java‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
package io.temporal.samples.workerversioning;
2+
3+
import org.slf4j.Logger;
4+
import org.slf4j.LoggerFactory;
5+
6+
public class ActivitiesImpl implements Activities {
7+
8+
private static final Logger logger = LoggerFactory.getLogger(ActivitiesImpl.class);
9+
10+
@Override
11+
public String someActivity(String calledBy) {
12+
logger.info("SomeActivity called by {}", calledBy);
13+
return "SomeActivity called by " + calledBy;
14+
}
15+
16+
@Override
17+
public String someIncompatibleActivity(IncompatibleActivityInput input) {
18+
logger.info(
19+
"SomeIncompatibleActivity called by {} with {}", input.getCalledBy(), input.getMoreData());
20+
return "SomeIncompatibleActivity called by "
21+
+ input.getCalledBy()
22+
+ " with "
23+
+ input.getMoreData();
24+
}
25+
}

‎core/src/main/java/io/temporal/samples/workerversioning/AutoUpgradingWorkflow.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package io.temporal.samples.workerversioning;
2+
3+
import io.temporal.workflow.SignalMethod;
4+
import io.temporal.workflow.WorkflowInterface;
5+
import io.temporal.workflow.WorkflowMethod;
6+
7+
@WorkflowInterface
8+
public interface AutoUpgradingWorkflow {
9+
10+
@WorkflowMethod
11+
void run();
12+
13+
@SignalMethod
14+
void doNextSignal(String signal);
15+
}

‎core/src/main/java/io/temporal/samples/workerversioning/AutoUpgradingWorkflowV1Impl.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
package io.temporal.samples.workerversioning;
2+
3+
import io.temporal.activity.ActivityOptions;
4+
import io.temporal.common.VersioningBehavior;
5+
import io.temporal.workflow.Workflow;
6+
import io.temporal.workflow.WorkflowVersioningBehavior;
7+
import java.time.Duration;
8+
import java.util.ArrayList;
9+
import java.util.List;
10+
import org.slf4j.Logger;
11+
12+
/**
13+
* This workflow will automatically move to the latest worker version. We'll be making changes to
14+
* it, which must be replay safe. Note that generally you won't want or need to include a version
15+
* number in your workflow name if you're using the worker versioning feature. This sample does it
16+
* to illustrate changes to the same code over time - but really what we're demonstrating here is
17+
* the evolution of what would have been one workflow definition.
18+
*/
19+
public class AutoUpgradingWorkflowV1Impl implements AutoUpgradingWorkflow {
20+
21+
private static final Logger logger = Workflow.getLogger(AutoUpgradingWorkflowV1Impl.class);
22+
23+
private final List<String> signals = new ArrayList<>();
24+
private final Activities activities =
25+
Workflow.newActivityStub(
26+
Activities.class,
27+
ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build());
28+
29+
@Override
30+
@WorkflowVersioningBehavior(VersioningBehavior.AUTO_UPGRADE)
31+
public void run() {
32+
logger.info("Changing workflow v1 started. StartTime: {}", Workflow.currentTimeMillis());
33+
34+
while (true) {
35+
Workflow.await(() -> !signals.isEmpty());
36+
String signal = signals.remove(0);
37+
38+
if ("do-activity".equals(signal)) {
39+
logger.info("Changing workflow v1 running activity");
40+
activities.someActivity("v1");
41+
} else {
42+
logger.info("Concluding workflow v1");
43+
return;
44+
}
45+
}
46+
}
47+
48+
@Override
49+
public void doNextSignal(String signal) {
50+
signals.add(signal);
51+
}
52+
}

‎core/src/main/java/io/temporal/samples/workerversioning/AutoUpgradingWorkflowV1bImpl.java‎

Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
package io.temporal.samples.workerversioning;
2+
3+
import io.temporal.activity.ActivityOptions;
4+
import io.temporal.common.VersioningBehavior;
5+
import io.temporal.workflow.Workflow;
6+
import io.temporal.workflow.WorkflowVersioningBehavior;
7+
import java.time.Duration;
8+
import java.util.ArrayList;
9+
import java.util.List;
10+
import org.slf4j.Logger;
11+
12+
/**
13+
* This represents us having made *compatible* changes to AutoUpgradingWorkflowV1Impl.
14+
*
15+
* <p>The compatible changes we've made are:
16+
*
17+
* <ul>
18+
* <li>Altering the log lines
19+
* <li>Using the `Workflow.getVersion` API to properly introduce branching behavior while
20+
* maintaining compatibility
21+
* </ul>
22+
*/
23+
public class AutoUpgradingWorkflowV1bImpl implements AutoUpgradingWorkflow {
24+
25+
private static final Logger logger = Workflow.getLogger(AutoUpgradingWorkflowV1bImpl.class);
26+
27+
private final List<String> signals = new ArrayList<>();
28+
private final Activities activities =
29+
Workflow.newActivityStub(
30+
Activities.class,
31+
ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build());
32+
33+
@Override
34+
@WorkflowVersioningBehavior(VersioningBehavior.AUTO_UPGRADE)
35+
public void run() {
36+
logger.info("Changing workflow v1b started. StartTime: {}", Workflow.currentTimeMillis());
37+
38+
while (true) {
39+
Workflow.await(() -> !signals.isEmpty());
40+
String signal = signals.remove(0);
41+
42+
if ("do-activity".equals(signal)) {
43+
logger.info("Changing workflow v1b running activity");
44+
int version = Workflow.getVersion("DifferentActivity", Workflow.DEFAULT_VERSION, 1);
45+
if (version == 1) {
46+
activities.someIncompatibleActivity(
47+
new Activities.IncompatibleActivityInput("v1b", "hello!"));
48+
} else {
49+
// Note it is a valid compatible change to alter the input to an activity.
50+
// However, because we're using the getVersion API, this branch will never be
51+
// taken.
52+
activities.someActivity("v1b");
53+
}
54+
} else {
55+
logger.info("Concluding workflow v1b");
56+
break;
57+
}
58+
}
59+
}
60+
61+
@Override
62+
public void doNextSignal(String signal) {
63+
signals.add(signal);
64+
}
65+
}

‎core/src/main/java/io/temporal/samples/workerversioning/PinnedWorkflow.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
package io.temporal.samples.workerversioning;
2+
3+
import io.temporal.workflow.SignalMethod;
4+
import io.temporal.workflow.WorkflowInterface;
5+
import io.temporal.workflow.WorkflowMethod;
6+
7+
@WorkflowInterface
8+
public interface PinnedWorkflow {
9+
10+
@WorkflowMethod
11+
void run();
12+
13+
@SignalMethod
14+
void doNextSignal(String signal);
15+
}

‎core/src/main/java/io/temporal/samples/workerversioning/PinnedWorkflowV1Impl.java‎

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package io.temporal.samples.workerversioning;
2+
3+
import io.temporal.activity.ActivityOptions;
4+
import io.temporal.common.VersioningBehavior;
5+
import io.temporal.workflow.Workflow;
6+
import io.temporal.workflow.WorkflowVersioningBehavior;
7+
import java.time.Duration;
8+
import java.util.ArrayList;
9+
import java.util.List;
10+
import org.slf4j.Logger;
11+
12+
/**
13+
* This workflow represents one that likely has a short lifetime, and we want to always stay pinned
14+
* to the same version it began on. Note that generally you won't want or need to include a version
15+
* number in your workflow name if you're using the worker versioning feature. This sample does it
16+
* to illustrate changes to the same code over time - but really what we're demonstrating here is
17+
* the evolution of what would have been one workflow definition.
18+
*/
19+
public class PinnedWorkflowV1Impl implements PinnedWorkflow {
20+
21+
private static final Logger logger = Workflow.getLogger(PinnedWorkflowV1Impl.class);
22+
23+
private final List<String> signals = new ArrayList<>();
24+
private final Activities activities =
25+
Workflow.newActivityStub(
26+
Activities.class,
27+
ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build());
28+
29+
@Override
30+
@WorkflowVersioningBehavior(VersioningBehavior.PINNED)
31+
public void run() {
32+
logger.info("Pinned Workflow v1 started. StartTime: {}", Workflow.currentTimeMillis());
33+
34+
while (true) {
35+
Workflow.await(() -> !signals.isEmpty());
36+
String signal = signals.remove(0);
37+
if ("conclude".equals(signal)) {
38+
break;
39+
}
40+
}
41+
42+
activities.someActivity("Pinned-v1");
43+
}
44+
45+
@Override
46+
public void doNextSignal(String signal) {
47+
signals.add(signal);
48+
}
49+
}

‎core/src/main/java/io/temporal/samples/workerversioning/PinnedWorkflowV2Impl.java‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
package io.temporal.samples.workerversioning;
2+
3+
import io.temporal.activity.ActivityOptions;
4+
import io.temporal.common.VersioningBehavior;
5+
import io.temporal.workflow.Workflow;
6+
import io.temporal.workflow.WorkflowVersioningBehavior;
7+
import java.time.Duration;
8+
import java.util.ArrayList;
9+
import java.util.List;
10+
import org.slf4j.Logger;
11+
12+
/**
13+
* This workflow has changes that would make it incompatible with v1, and aren't protected by a
14+
* patch.
15+
*/
16+
public class PinnedWorkflowV2Impl implements PinnedWorkflow {
17+
18+
private static final Logger logger = Workflow.getLogger(PinnedWorkflowV2Impl.class);
19+
20+
private final List<String> signals = new ArrayList<>();
21+
private final Activities activities =
22+
Workflow.newActivityStub(
23+
Activities.class,
24+
ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(10)).build());
25+
26+
@Override
27+
@WorkflowVersioningBehavior(VersioningBehavior.PINNED)
28+
public void run() {
29+
logger.info("Pinned Workflow v2 started. StartTime: {}", Workflow.currentTimeMillis());
30+
31+
// Here we call an activity where we didn't before, which is an incompatible change.
32+
activities.someActivity("Pinned-v2");
33+
34+
while (true) {
35+
Workflow.await(() -> !signals.isEmpty());
36+
String signal = signals.remove(0);
37+
if ("conclude".equals(signal)) {
38+
break;
39+
}
40+
}
41+
42+
// We've also changed the activity type here, another incompatible change
43+
activities.someIncompatibleActivity(
44+
new Activities.IncompatibleActivityInput("Pinned-v2", "hi"));
45+
}
46+
47+
@Override
48+
public void doNextSignal(String signal) {
49+
signals.add(signal);
50+
}
51+
}

0 commit comments

Comments
 (0)