[wip] auto heartbeater sample

Signed-off-by: Tihomir Surdilovic <tihomir@temporal.io>

Author: tsurdilo

core/src/main/java/io/temporal/samples/autoheartbeat/AutoActivities.java

@@ -0,0 +1,29 @@
+/*
+ *  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.autoheartbeat;
+
+import io.temporal.activity.ActivityInterface;
+
+@ActivityInterface
+public interface AutoActivities {
+  String runActivityOne(String input);
+
+  String runActivityTwo(String input);
+}

core/src/main/java/io/temporal/samples/autoheartbeat/AutoActivitiesImpl.java

@@ -0,0 +1,87 @@
+/*
+ *  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.autoheartbeat;
+
+import io.temporal.activity.Activity;
+import io.temporal.client.ActivityCompletionException;
+import java.util.concurrent.TimeUnit;
+
+public class AutoActivitiesImpl implements AutoActivities {
+
+  @Override
+  public String runActivityOne(String input) {
+    return runActivity("runActivityOne - " + input);
+  }
+
+  @Override
+  public String runActivityTwo(String input) {
+    return runActivity("runActivityTwo - " + input);
+  }
+
+  @SuppressWarnings("FutureReturnValueIgnored")
+  private String runActivity(String input) {
+    // Calculate heartbeat period based on our heartbeat timeout
+    // Start Autoheartbeater
+    AutoHeartbeater autoHearbeater =
+        new AutoHeartbeater(
+            getHeartbeatPeriod(), 0, TimeUnit.SECONDS, Activity.getExecutionContext(), input);
+    autoHearbeater.start();
+
+    // For sample our activity just sleeps for a second for 20 seconds
+    for (int i = 0; i < 20; i++) {
+      try {
+        sleep(1);
+      } catch (ActivityCompletionException e) {
+        System.out.println(
+            "Activity type:"
+                + e.getActivityType().get()
+                + "Activiy id: "
+                + e.getActivityId().get()
+                + "Workflow id: "
+                + e.getWorkflowId().get()
+                + "Workflow runid: "
+                + e.getRunId().get()
+                + " was canceled. Shutting down auto heartbeats");
+        autoHearbeater.stop();
+        // We want to rethrow the cancel failure
+        throw e;
+      }
+    }
+    return "Activity completed: " + input;
+  }
+
+  private long getHeartbeatPeriod() {
+    // Note you can add checks if heartbeat timeout is set if not and
+    // decide to log / fail activity / not start autoheartbeater based on your business logic
+
+    // For sample we want to heartbeat 1 seconds less than heartbeat timeout
+    return Activity.getExecutionContext().getInfo().getHeartbeatTimeout().getSeconds() <= 1
+        ? 1
+        : Activity.getExecutionContext().getInfo().getHeartbeatTimeout().getSeconds() - 1;
+  }
+
+  private void sleep(int seconds) {
+    try {
+      Thread.sleep(TimeUnit.SECONDS.toMillis(seconds));
+    } catch (InterruptedException ee) {
+      // Empty
+    }
+  }
+}

core/src/main/java/io/temporal/samples/autoheartbeat/AutoHeartbeater.java

@@ -0,0 +1,89 @@
+/*
+ *  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.autoheartbeat;
+
+import io.temporal.activity.ActivityExecutionContext;
+import java.time.Instant;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+public class AutoHeartbeater {
+  private final long period;
+  private final long initialDelay;
+  private final TimeUnit periodTimeUnit;
+  private final ScheduledExecutorService timerService =
+      Executors.newSingleThreadScheduledExecutor();
+  private final ActivityExecutionContext context;
+  private final Object details;
+  private String heartbeaterId;
+
+  public AutoHeartbeater(
+      long period,
+      long initialDelay,
+      TimeUnit periodTimeUnit,
+      ActivityExecutionContext context,
+      Object details) {
+    this.period = period;
+    this.initialDelay = initialDelay;
+    this.periodTimeUnit = periodTimeUnit;
+    this.context = context;
+    this.details = details;
+    // Set to activity id better, for sample we just use type
+    heartbeaterId = context.getInfo().getActivityType();
+  }
+
+  public ScheduledFuture<?> start() {
+    System.out.println("Autoheartbeater[" + heartbeaterId + "] starting...");
+    return timerService.scheduleAtFixedRate(
+        () -> {
+          try {
+            System.out.println(
+                "Autoheartbeater["
+                    + heartbeaterId
+                    + "]"
+                    + "heartbeating at: "
+                    + printShortCurrentTime());
+            context.heartbeat(details);
+          } catch (Exception e) {
+            System.out.println("Stopping: " + e.getMessage());
+            stop();
+          }
+        },
+        initialDelay,
+        period,
+        periodTimeUnit);
+  }
+
+  public void stop() {
+    System.out.println("Autoheartbeater being requested to stop.");
+    // Try not to execute another heartbeat that could have been queued up
+    timerService.shutdownNow();
+  }
+
+  private String printShortCurrentTime() {
+    return DateTimeFormatter.ofPattern("HH:mm:ss")
+        .withZone(ZoneId.systemDefault())
+        .format(Instant.now());
+  }
+}

core/src/main/java/io/temporal/samples/autoheartbeat/AutoWorkflow.java

@@ -0,0 +1,33 @@
+/*
+ *  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.autoheartbeat;
+
+import io.temporal.workflow.SignalMethod;
+import io.temporal.workflow.WorkflowInterface;
+import io.temporal.workflow.WorkflowMethod;
+
+@WorkflowInterface
+public interface AutoWorkflow {
+  @WorkflowMethod
+  String exec(String input);
+
+  @SignalMethod
+  void cancelActivity();
+}

core/src/main/java/io/temporal/samples/autoheartbeat/AutoWorkflowImpl.java

@@ -0,0 +1,89 @@
+/*
+ *  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.autoheartbeat;
+
+import io.temporal.activity.ActivityOptions;
+import io.temporal.failure.ActivityFailure;
+import io.temporal.failure.CanceledFailure;
+import io.temporal.workflow.Async;
+import io.temporal.workflow.CancellationScope;
+import io.temporal.workflow.Promise;
+import io.temporal.workflow.Workflow;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+public class AutoWorkflowImpl implements AutoWorkflow {
+  private CancellationScope scope;
+
+  @Override
+  public String exec(String input) {
+    // Crete separate workflow stubs for same interface so we can show
+    // use of different heartbeat timeouts
+    AutoActivities activitiesOne =
+        Workflow.newActivityStub(
+            AutoActivities.class,
+            ActivityOptions.newBuilder()
+                .setStartToCloseTimeout(Duration.ofSeconds(22))
+                .setHeartbeatTimeout(Duration.ofSeconds(3))
+                .build());
+
+    AutoActivities activitiesTwo =
+        Workflow.newActivityStub(
+            AutoActivities.class,
+            ActivityOptions.newBuilder()
+                .setStartToCloseTimeout(Duration.ofSeconds(22))
+                .setHeartbeatTimeout(Duration.ofSeconds(5))
+                .build());
+
+    // Start our activity in CancellationScope so we can cancel it if needed
+    List<Promise<String>> activityPromises = new ArrayList<>();
+    scope =
+        Workflow.newCancellationScope(
+            () -> {
+              activityPromises.add(Async.function(activitiesOne::runActivityOne, input));
+              activityPromises.add(Async.function(activitiesTwo::runActivityTwo, input));
+            });
+
+    scope.run();
+
+    try {
+      Promise.allOf(activityPromises).get();
+      String result = "";
+      for (Promise<String> pr : activityPromises) {
+        result += pr.get();
+      }
+      return result;
+    } catch (ActivityFailure e) {
+      if (e.getCause() instanceof CanceledFailure) {
+        // We dont want workflow to fail in we canceled our scope, just log and return
+        return "Workflow result after activity cancellation";
+      } else {
+        // We want to fail execution on any other failures except cancellation
+        throw e;
+      }
+    }
+  }
+
+  @Override
+  public void cancelActivity() {
+    scope.cancel("Canceling scope from signal handler");
+  }
+}

core/src/main/java/io/temporal/samples/autoheartbeat/Starter.java

@@ -0,0 +1,58 @@
+/*
+ *  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.autoheartbeat;
+
+import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowOptions;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import io.temporal.worker.Worker;
+import io.temporal.worker.WorkerFactory;
+
+public class Starter {
+  static final String TASK_QUEUE = "AutoheartbeatTaskQueue";
+  static final String WORKFLOW_ID = "AutoHeartbeatWorkflow";
+
+  public static void main(String[] args) {
+    WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs();
+    WorkflowClient client = WorkflowClient.newInstance(service);
+    WorkerFactory factory = WorkerFactory.newInstance(client);
+    Worker worker = factory.newWorker(TASK_QUEUE);
+
+    worker.registerWorkflowImplementationTypes(AutoWorkflowImpl.class);
+    worker.registerActivitiesImplementations(new AutoActivitiesImpl());
+
+    factory.start();
+
+    AutoWorkflow workflow =
+        client.newWorkflowStub(
+            AutoWorkflow.class,
+            WorkflowOptions.newBuilder()
+                .setWorkflowId(WORKFLOW_ID)
+                .setTaskQueue(TASK_QUEUE)
+                .build());
+
+    try {
+      String result = workflow.exec("Auto heartbeating is cool");
+      System.out.println("Result: " + result);
+    } catch (Exception e) {
+      System.out.println("Workflow exec exception: " + e.getClass().getName());
+    }
+  }
+}