From ca8ea4c833bec9f3f5f0cad18e0cb38a9ff6f0fb Mon Sep 17 00:00:00 2001 From: kaibocai <89094811+kaibocai@users.noreply.github.com> Date: Mon, 4 Apr 2022 12:09:17 -0500 Subject: [PATCH 1/3] termination support - add one test case --- .../durabletask/DurableTaskClient.java | 6 ++++ .../durabletask/DurableTaskGrpcClient.java | 14 +++++++++ .../TaskOrchestrationExecutor.java | 10 +++++-- .../durabletask/IntegrationTests.java | 29 +++++++++++++++++-- 4 files changed, 55 insertions(+), 4 deletions(-) diff --git a/sdk/src/main/java/com/microsoft/durabletask/DurableTaskClient.java b/sdk/src/main/java/com/microsoft/durabletask/DurableTaskClient.java index 8bdf768..efaed55 100644 --- a/sdk/src/main/java/com/microsoft/durabletask/DurableTaskClient.java +++ b/sdk/src/main/java/com/microsoft/durabletask/DurableTaskClient.java @@ -61,4 +61,10 @@ public OrchestrationMetadata waitForInstanceCompletion(String instanceId, Durati } public abstract OrchestrationMetadata waitForInstanceCompletion(String instanceId, Duration timeout, boolean getInputsAndOutputs); + + public void terminate(String instanceId){ + this.terminate(instanceId, null); + } + + public abstract void terminate(String instanceId, Object output); } \ No newline at end of file diff --git a/sdk/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java b/sdk/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java index de31738..f3e86d0 100644 --- a/sdk/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java +++ b/sdk/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java @@ -15,10 +15,12 @@ import java.util.Objects; import java.util.UUID; import java.util.concurrent.TimeUnit; +import java.util.logging.Logger; // TODO: Create async flavors of the public APIs that call into the sidecar public class DurableTaskGrpcClient extends DurableTaskClient { private static final int DEFAULT_PORT = 4001; + private static final Logger logger = Logger.getLogger(DurableTaskGrpcClient.class.getPackage().getName()); private final DataConverter dataConverter; private final ManagedChannel managedSidecarChannel; @@ -167,6 +169,18 @@ public OrchestrationMetadata waitForInstanceCompletion(String instanceId, Durati return new OrchestrationMetadata(response, this.dataConverter, request.getGetInputsAndOutputs()); } + @Override + public void terminate(String instanceId, Object output) { + Helpers.throwIfArgumentNull(instanceId, "instanceId"); + String serializeOutput = dataConverter.serialize(output); + this.logger.fine(() -> String.format( + "Terminating instance %s with reason: %s", + instanceId, + serializeOutput != null ? serializeOutput : "(null)")); + TerminateRequest.Builder builder = TerminateRequest.newBuilder().setInstanceId(instanceId).setOutput(StringValue.of(serializeOutput)); + this.sidecarClient.terminateInstance(builder.build()); + } + public static Builder newBuilder() { return new Builder(); } diff --git a/sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java b/sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java index 7e90253..32204eb 100644 --- a/sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java +++ b/sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java @@ -618,6 +618,11 @@ private void handleSubOrchestrationFailed(HistoryEvent e){ task.completeExceptionally(exception); } + private void handleExecutionTerminated(HistoryEvent e) { + ExecutionTerminatedEvent executionTerminatedEvent = e.getExecutionTerminated(); + this.completeInternal(executionTerminatedEvent.getInput().getValue(), null, OrchestrationStatus.ORCHESTRATION_STATUS_TERMINATED); + } + @Override public void complete(Object output) { if (this.continuedAsNew) { @@ -705,8 +710,9 @@ private void processEvent(HistoryEvent e) throws TaskFailedException, Orchestrat // break; // case EXECUTIONFAILED: // break; -// case EXECUTIONTERMINATED: -// break; + case EXECUTIONTERMINATED: + this.handleExecutionTerminated(e); + break; case TASKSCHEDULED: this.handleTaskScheduled(e); break; diff --git a/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java b/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java index 798874c..e8e698d 100644 --- a/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java +++ b/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java @@ -269,7 +269,7 @@ void continueAsNew(){ } @Test - void continueAsNewWithExternalEvents(){ + void continueAsNewWithExternalEvents() { final String orchestratorName = "continueAsNewWithExternalEvents"; final String eventName = "MyEvent"; final int expectedEventCount = 10; @@ -285,7 +285,7 @@ void continueAsNewWithExternalEvents(){ } }).buildAndStart(); DurableTaskClient client = DurableTaskGrpcClient.newBuilder().build(); - try(worker; client){ + try (worker; client) { String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName, 0); for (int i = 0; i < expectedEventCount; i++) { @@ -299,6 +299,31 @@ void continueAsNewWithExternalEvents(){ } } + @Test + void termination() { + final String orchestratorName = "Termination"; + final Duration delay = Duration.ofSeconds(3); + + DurableTaskGrpcWorker worker = this.createWorkerBuilder() + .addOrchestrator(orchestratorName, ctx -> ctx.createTimer(delay).get()) + .buildAndStart(); + + DurableTaskClient client = DurableTaskGrpcClient.newBuilder().build(); + try (worker; client) { + String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName); + String expectOutput = "I'll be back."; + String expectOutputWithQuote = new StringBuilder("\"\\\"").append(expectOutput).append("\\\"\"").toString(); + client.terminate(instanceId, expectOutput); + OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true); + + assertNotNull(instance); + System.out.println(instance.getSerializedOutput()); + assertEquals(instanceId, instance.getInstanceId()); + assertEquals(OrchestrationRuntimeStatus.TERMINATED, instance.getRuntimeStatus()); + assertEquals(expectOutputWithQuote, instance.getSerializedOutput()); + } + } + @Test void activityFanOut() throws IOException { final String orchestratorName = "ActivityFanOut"; From 3af962b3413227305e4a2dd4fad14f3b076d6983 Mon Sep 17 00:00:00 2001 From: kaibocai <89094811+kaibocai@users.noreply.github.com> Date: Tue, 5 Apr 2022 16:32:38 -0500 Subject: [PATCH 2/3] resolve comments - refactor the test case --- .../durabletask/DurableTaskClient.java | 3 ++- .../durabletask/DurableTaskGrpcClient.java | 7 ++++--- .../durabletask/IntegrationTests.java | 20 +++++++++++++++---- 3 files changed, 22 insertions(+), 8 deletions(-) diff --git a/sdk/src/main/java/com/microsoft/durabletask/DurableTaskClient.java b/sdk/src/main/java/com/microsoft/durabletask/DurableTaskClient.java index efaed55..67bce77 100644 --- a/sdk/src/main/java/com/microsoft/durabletask/DurableTaskClient.java +++ b/sdk/src/main/java/com/microsoft/durabletask/DurableTaskClient.java @@ -2,6 +2,7 @@ // Licensed under the MIT License. package com.microsoft.durabletask; +import javax.annotation.Nullable; import java.time.Duration; // TODO: JavaDocs @@ -66,5 +67,5 @@ public void terminate(String instanceId){ this.terminate(instanceId, null); } - public abstract void terminate(String instanceId, Object output); + public abstract void terminate(String instanceId, @Nullable Object output); } \ No newline at end of file diff --git a/sdk/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java b/sdk/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java index f3e86d0..e27ebca 100644 --- a/sdk/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java +++ b/sdk/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java @@ -10,6 +10,7 @@ import io.grpc.*; +import javax.annotation.Nullable; import java.time.Duration; import java.time.Instant; import java.util.Objects; @@ -170,11 +171,11 @@ public OrchestrationMetadata waitForInstanceCompletion(String instanceId, Durati } @Override - public void terminate(String instanceId, Object output) { + public void terminate(String instanceId, @Nullable Object output) { Helpers.throwIfArgumentNull(instanceId, "instanceId"); - String serializeOutput = dataConverter.serialize(output); + String serializeOutput = this.dataConverter.serialize(output); this.logger.fine(() -> String.format( - "Terminating instance %s with reason: %s", + "Terminating instance %s and setting output to: %s", instanceId, serializeOutput != null ? serializeOutput : "(null)")); TerminateRequest.Builder builder = TerminateRequest.newBuilder().setInstanceId(instanceId).setOutput(StringValue.of(serializeOutput)); diff --git a/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java b/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java index e8e698d..9fb0d63 100644 --- a/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java +++ b/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java @@ -312,15 +312,27 @@ void termination() { try (worker; client) { String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName); String expectOutput = "I'll be back."; - String expectOutputWithQuote = new StringBuilder("\"\\\"").append(expectOutput).append("\\\"\"").toString(); - client.terminate(instanceId, expectOutput); + Output testOutput = new IntegrationTests().new Output(); + testOutput.setQuote(expectOutput); + client.terminate(instanceId, testOutput); OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true); - assertNotNull(instance); System.out.println(instance.getSerializedOutput()); assertEquals(instanceId, instance.getInstanceId()); assertEquals(OrchestrationRuntimeStatus.TERMINATED, instance.getRuntimeStatus()); - assertEquals(expectOutputWithQuote, instance.getSerializedOutput()); + assertEquals(expectOutput, instance.readOutputAs(Output.class).getQuote()); + } + } + + class Output{ + String quote; + + public String getQuote() { + return quote; + } + + public void setQuote(String quote) { + this.quote = quote; } } From 04dd086798148bac3c716065c1b3168ecb700f83 Mon Sep 17 00:00:00 2001 From: Chris Gillum Date: Tue, 5 Apr 2022 17:20:17 -0700 Subject: [PATCH 3/3] Fixed double-serialization issue --- .../TaskOrchestrationExecutor.java | 19 +++++++++++++------ .../durabletask/IntegrationTests.java | 18 ++---------------- 2 files changed, 15 insertions(+), 22 deletions(-) diff --git a/sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java b/sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java index 32204eb..acba366 100644 --- a/sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java +++ b/sdk/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java @@ -626,9 +626,9 @@ private void handleExecutionTerminated(HistoryEvent e) { @Override public void complete(Object output) { if (this.continuedAsNew) { - this.completeInternal(this.continuedAsNewInput, null, OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW); + this.completeInternal(this.continuedAsNewInput, OrchestrationStatus.ORCHESTRATION_STATUS_CONTINUED_AS_NEW); } else { - this.completeInternal(output, null, OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); + this.completeInternal(output, OrchestrationStatus.ORCHESTRATION_STATUS_COMPLETED); } } @@ -638,16 +638,23 @@ public void fail(FailureDetails failureDetails) { this.completeInternal(null, failureDetails, OrchestrationStatus.ORCHESTRATION_STATUS_FAILED); } - private void completeInternal(Object output, FailureDetails failureDetails, OrchestrationStatus runtimeStatus) { + private void completeInternal(Object output, OrchestrationStatus runtimeStatus) { + String resultAsJson = TaskOrchestrationExecutor.this.dataConverter.serialize(output); + this.completeInternal(resultAsJson, null, runtimeStatus); + } + + private void completeInternal( + @Nullable String rawOutput, + @Nullable FailureDetails failureDetails, + OrchestrationStatus runtimeStatus) { Helpers.throwIfOrchestratorComplete(this.isComplete); int id = this.sequenceNumber++; CompleteOrchestrationAction.Builder builder = CompleteOrchestrationAction.newBuilder(); builder.setOrchestrationStatus(runtimeStatus); - if (output != null) { - String resultAsJson = TaskOrchestrationExecutor.this.dataConverter.serialize(output); - builder.setResult(StringValue.of(resultAsJson)); + if (rawOutput != null) { + builder.setResult(StringValue.of(rawOutput)); } if (failureDetails != null) { diff --git a/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java b/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java index 9fb0d63..2c8ab87 100644 --- a/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java +++ b/sdk/src/test/java/com/microsoft/durabletask/IntegrationTests.java @@ -312,27 +312,13 @@ void termination() { try (worker; client) { String instanceId = client.scheduleNewOrchestrationInstance(orchestratorName); String expectOutput = "I'll be back."; - Output testOutput = new IntegrationTests().new Output(); - testOutput.setQuote(expectOutput); - client.terminate(instanceId, testOutput); + client.terminate(instanceId, expectOutput); OrchestrationMetadata instance = client.waitForInstanceCompletion(instanceId, defaultTimeout, true); assertNotNull(instance); System.out.println(instance.getSerializedOutput()); assertEquals(instanceId, instance.getInstanceId()); assertEquals(OrchestrationRuntimeStatus.TERMINATED, instance.getRuntimeStatus()); - assertEquals(expectOutput, instance.readOutputAs(Output.class).getQuote()); - } - } - - class Output{ - String quote; - - public String getQuote() { - return quote; - } - - public void setQuote(String quote) { - this.quote = quote; + assertEquals(expectOutput, instance.readOutputAs(String.class)); } }