-
Notifications
You must be signed in to change notification settings - Fork 148
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
EarlyReturn sample (Update-With-Start) #689
Changes from 9 commits
284f4fa
5082bb6
d38125b
86c4883
497f2dc
d161cbd
7700fa1
463c2c7
78bd0c2
b8513d9
9037387
cb93bfb
f40e832
ce6cc56
8095cc7
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,90 @@ | ||
/* | ||
* Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved | ||
* | ||
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
* | ||
* Modifications copyright (C) 2017 Uber Technologies, Inc. | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"). You may not | ||
* use this file except in compliance with the License. A copy of the License is | ||
* located at | ||
* | ||
* http://aws.amazon.com/apache2.0 | ||
* | ||
* or in the "license" file accompanying this file. This file is distributed on | ||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either | ||
* express or implied. See the License for the specific language governing | ||
* permissions and limitations under the License. | ||
*/ | ||
|
||
package io.temporal.samples.earlyreturn; | ||
|
||
import io.temporal.client.*; | ||
import io.temporal.serviceclient.WorkflowServiceStubs; | ||
|
||
public class EarlyReturnClient { | ||
private static final String TASK_QUEUE = "EarlyReturnTaskQueue"; | ||
private static final String WORKFLOW_ID_PREFIX = "early-return-workflow-"; | ||
|
||
public static void main(String[] args) { | ||
WorkflowClient client = setupWorkflowClient(); | ||
runWorkflowWithUpdateWithStart(client); | ||
} | ||
|
||
// Set up the WorkflowClient | ||
public static WorkflowClient setupWorkflowClient() { | ||
WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs(); | ||
return WorkflowClient.newInstance(service); | ||
} | ||
|
||
// Run workflow using 'updateWithStart' | ||
private static void runWorkflowWithUpdateWithStart(WorkflowClient client) { | ||
TransactionRequest txRequest = | ||
new TransactionRequest( | ||
"Bob", "Alice", | ||
-1000); // Change this amount to a negative number to have initTransaction fail | ||
|
||
WorkflowOptions options = buildWorkflowOptions(); | ||
TransactionWorkflow workflow = client.newWorkflowStub(TransactionWorkflow.class, options); | ||
|
||
System.out.println("Starting workflow with UpdateWithStart"); | ||
|
||
UpdateWithStartWorkflowOperation<TxResult> updateOp = | ||
UpdateWithStartWorkflowOperation.newBuilder(workflow::returnInitResult) | ||
.setWaitForStage(WorkflowUpdateStage.COMPLETED) // Wait for update to complete | ||
.build(); | ||
|
||
TxResult updateResult = null; | ||
try { | ||
WorkflowUpdateHandle<TxResult> updateHandle = | ||
WorkflowClient.updateWithStart(workflow::processTransaction, txRequest, updateOp); | ||
|
||
updateResult = updateHandle.getResultAsync().get(); | ||
|
||
System.out.println( | ||
"Workflow initialized with result: " | ||
+ updateResult.getStatus() | ||
+ " (transactionId: " | ||
+ updateResult.getTransactionId() | ||
+ ")"); | ||
|
||
TxResult result = WorkflowStub.fromTyped(workflow).getResult(TxResult.class); | ||
System.out.println( | ||
"Workflow completed with result: " | ||
+ result.getStatus() | ||
+ " (transactionId: " | ||
+ result.getTransactionId() | ||
+ ")"); | ||
} catch (Exception e) { | ||
System.err.println("Transaction initialization failed: " + e.getMessage()); | ||
} | ||
} | ||
|
||
// Build WorkflowOptions with task queue and unique ID | ||
private static WorkflowOptions buildWorkflowOptions() { | ||
return WorkflowOptions.newBuilder() | ||
.setTaskQueue(TASK_QUEUE) | ||
.setWorkflowId(WORKFLOW_ID_PREFIX + System.currentTimeMillis()) | ||
.build(); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,44 @@ | ||
/* | ||
* Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved | ||
* | ||
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
* | ||
* Modifications copyright (C) 2017 Uber Technologies, Inc. | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"). You may not | ||
* use this file except in compliance with the License. A copy of the License is | ||
* located at | ||
* | ||
* http://aws.amazon.com/apache2.0 | ||
* | ||
* or in the "license" file accompanying this file. This file is distributed on | ||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either | ||
* express or implied. See the License for the specific language governing | ||
* permissions and limitations under the License. | ||
*/ | ||
|
||
package io.temporal.samples.earlyreturn; | ||
|
||
import io.temporal.client.WorkflowClient; | ||
import io.temporal.worker.Worker; | ||
import io.temporal.worker.WorkerFactory; | ||
|
||
public class EarlyReturnWorker { | ||
private static final String TASK_QUEUE = "EarlyReturnTaskQueue"; | ||
|
||
public static void main(String[] args) { | ||
WorkflowClient client = EarlyReturnClient.setupWorkflowClient(); | ||
startWorker(client); | ||
} | ||
|
||
private static void startWorker(WorkflowClient client) { | ||
WorkerFactory factory = WorkerFactory.newInstance(client); | ||
Worker worker = factory.newWorker(TASK_QUEUE); | ||
|
||
worker.registerWorkflowImplementationTypes(TransactionWorkflowImpl.class); | ||
worker.registerActivitiesImplementations(new TransactionActivitiesImpl()); | ||
|
||
factory.start(); | ||
System.out.println("Worker started"); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,24 @@ | ||
### Early-Return Sample | ||
|
||
This sample demonstrates an early-return from a workflow. | ||
|
||
By utilizing Update-with-Start, a client can start a new workflow and synchronously receive | ||
a response mid-workflow, while the workflow continues to run to completion. | ||
|
||
To run the sample, start the worker: | ||
```bash | ||
./gradlew -q execute -PmainClass=io.temporal.samples.earlyreturn.EarlyReturnWorker | ||
``` | ||
|
||
Then, start the client: | ||
|
||
```bash | ||
./gradlew -q execute -PmainClass=io.temporal.samples.earlyreturn.EarlyReturnClient | ||
``` | ||
|
||
* The client will start a workflow using Update-With-Start. | ||
* Update-With-Start will trigger an initialization step. | ||
* If the initialization step succeeds (default), intialization will return to the client with a transaction ID and the workflow will continue. The workflow will then complete and return the final result. | ||
* If the intitialization step fails (amount <= 0), the workflow will return to the client with an error message and the workflow will run an activity to cancel the transaction. | ||
|
||
To trigger a failed initialization, set the amount to <= 0 in the `EarlyReturnClient` class's `runWorkflowWithUpdateWithStart` method and re-run the client. |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,69 @@ | ||
/* | ||
* Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved | ||
* | ||
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
* | ||
* Modifications copyright (C) 2017 Uber Technologies, Inc. | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"). You may not | ||
* use this file except in compliance with the License. A copy of the License is | ||
* located at | ||
* | ||
* http://aws.amazon.com/apache2.0 | ||
* | ||
* or in the "license" file accompanying this file. This file is distributed on | ||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either | ||
* express or implied. See the License for the specific language governing | ||
* permissions and limitations under the License. | ||
*/ | ||
|
||
package io.temporal.samples.earlyreturn; | ||
|
||
import com.fasterxml.jackson.annotation.JsonCreator; | ||
import com.fasterxml.jackson.annotation.JsonProperty; | ||
|
||
public final class Transaction { | ||
private final String id; | ||
private final String sourceAccount; | ||
private final String targetAccount; | ||
private final int amount; | ||
|
||
@JsonCreator(mode = JsonCreator.Mode.PROPERTIES) | ||
public Transaction( | ||
@JsonProperty("id") String id, | ||
@JsonProperty("sourceAccount") String sourceAccount, | ||
@JsonProperty("targetAccount") String targetAccount, | ||
@JsonProperty("amount") int amount) { | ||
this.id = id; | ||
this.sourceAccount = sourceAccount; | ||
this.targetAccount = targetAccount; | ||
this.amount = amount; | ||
} | ||
|
||
@JsonProperty("id") | ||
public String getId() { | ||
return id; | ||
} | ||
|
||
@JsonProperty("sourceAccount") | ||
public String getSourceAccount() { | ||
return sourceAccount; | ||
} | ||
|
||
@JsonProperty("targetAccount") | ||
public String getTargetAccount() { | ||
return targetAccount; | ||
} | ||
|
||
@JsonProperty("amount") | ||
public int getAmount() { | ||
return amount; | ||
} | ||
|
||
@Override | ||
public String toString() { | ||
return String.format( | ||
"Transaction{id='%s', sourceAccount='%s', targetAccount='%s', amount=%d}", | ||
id, sourceAccount, targetAccount, amount); | ||
} | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,35 @@ | ||
/* | ||
* Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved | ||
* | ||
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
* | ||
* Modifications copyright (C) 2017 Uber Technologies, Inc. | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"). You may not | ||
* use this file except in compliance with the License. A copy of the License is | ||
* located at | ||
* | ||
* http://aws.amazon.com/apache2.0 | ||
* | ||
* or in the "license" file accompanying this file. This file is distributed on | ||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either | ||
* express or implied. See the License for the specific language governing | ||
* permissions and limitations under the License. | ||
*/ | ||
|
||
package io.temporal.samples.earlyreturn; | ||
|
||
import io.temporal.activity.ActivityInterface; | ||
import io.temporal.activity.ActivityMethod; | ||
|
||
@ActivityInterface | ||
public interface TransactionActivities { | ||
@ActivityMethod | ||
Transaction initTransaction(TransactionRequest tx); | ||
steveandroulakis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
|
||
@ActivityMethod | ||
void cancelTransaction(TransactionRequest txRequest); | ||
|
||
@ActivityMethod | ||
void completeTransaction(Transaction tx); | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,75 @@ | ||
/* | ||
* Copyright (c) 2020 Temporal Technologies, Inc. All Rights Reserved | ||
* | ||
* Copyright 2012-2016 Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
* | ||
* Modifications copyright (C) 2017 Uber Technologies, Inc. | ||
* | ||
* Licensed under the Apache License, Version 2.0 (the "License"). You may not | ||
* use this file except in compliance with the License. A copy of the License is | ||
* located at | ||
* | ||
* http://aws.amazon.com/apache2.0 | ||
* | ||
* or in the "license" file accompanying this file. This file is distributed on | ||
* an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either | ||
* express or implied. See the License for the specific language governing | ||
* permissions and limitations under the License. | ||
*/ | ||
|
||
package io.temporal.samples.earlyreturn; | ||
|
||
import io.temporal.failure.ApplicationFailure; | ||
|
||
public class TransactionActivitiesImpl implements TransactionActivities { | ||
|
||
@Override | ||
public Transaction initTransaction(TransactionRequest txRequest) { | ||
System.out.println("Initializing transaction"); | ||
sleep(500); | ||
if (txRequest.getAmount() <= 0) { | ||
System.out.println("Invalid amount: " + txRequest.getAmount()); | ||
throw ApplicationFailure.newNonRetryableFailure( | ||
"Non-retryable Activity Failure: Invalid Amount", "InvalidAmount"); | ||
} | ||
// Simulate transaction ID generation | ||
String txId = "TXID" + String.format("%010d", (long) (Math.random() * 1_000_000_0000L)); | ||
Transaction tx = | ||
new Transaction( | ||
txId, | ||
txRequest.getSourceAccount(), | ||
txRequest.getTargetAccount(), | ||
txRequest.getAmount()); | ||
|
||
sleep(500); | ||
steveandroulakis marked this conversation as resolved.
Show resolved
Hide resolved
|
||
return tx; | ||
} | ||
|
||
@Override | ||
public void cancelTransaction(TransactionRequest txRequest) { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I think There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I'm diverging more and more from the original Go sample (which is fine). Right now, I think of an initialization as something that results in the 'minting' of a Transaction ID. And a failure of initialization causes an activity failure which triggers a cancellation. I can't think of a way of having
Thinking blank-slate for a moment, and about the purpose of an "Early Return" type sample. @Quinn-With-Two-Ns -- What do you think you would want to see from a scenario where a transaction fails on initialization (e.g. the workflow update with start)? What data should be returned and what compensation should take place after? I don't think I should spend any more hours chewing on this sample without figuring that out first. Thanks! There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Yeah you raise a great point. I think I would separate the creation of the ID from the |
||
System.out.println("Cancelling transaction"); | ||
sleep(300); | ||
System.out.println("Transaction cancelled"); | ||
} | ||
|
||
@Override | ||
public void completeTransaction(Transaction tx) { | ||
System.out.println( | ||
"Sending $" | ||
+ tx.getAmount() | ||
+ " from " | ||
+ tx.getSourceAccount() | ||
+ " to " | ||
+ tx.getTargetAccount()); | ||
sleep(2000); | ||
System.out.println("Transaction completed successfully"); | ||
} | ||
|
||
private void sleep(long millis) { | ||
try { | ||
Thread.sleep(millis); | ||
} catch (InterruptedException e) { | ||
Thread.currentThread().interrupt(); | ||
} | ||
} | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In the Go example, I'm generated the transaction ID on the client and use it in the Workflow ID. That way, the user would receive an error when they start a workflow for the same transaction again. I don't think it's strictly necessary to mimic that here, but I thought I'd mention it.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Gotcha. I intentionally changed it so the early-return would contain meaningful data usable for the rest of the transaction (an early-return / initialization step that returns null isn't very illustrative of real-world value imo)
I plan on following Quinn's suggestion here of still having the ID minted as part of an activity, but doing it in a separate one as part of the initialization step #689 (comment) -- Let me know if there's anything you'd do differently