Add custom annotation sample

Author: Quinn-With-Two-Ns

core/src/main/java/io/temporal/samples/customannotation/CustomAnnotation.java

@@ -0,0 +1,226 @@
+/*
+ *  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.customannotation;
+
+import io.temporal.activity.ActivityInterface;
+import io.temporal.activity.ActivityOptions;
+import io.temporal.client.WorkflowClient;
+import io.temporal.client.WorkflowOptions;
+import io.temporal.common.RetryOptions;
+import io.temporal.serviceclient.WorkflowServiceStubs;
+import io.temporal.worker.Worker;
+import io.temporal.worker.WorkerFactory;
+import io.temporal.worker.WorkerFactoryOptions;
+import io.temporal.workflow.Workflow;
+import io.temporal.workflow.WorkflowInterface;
+import io.temporal.workflow.WorkflowMethod;
+import java.time.Duration;
+
+public class CustomAnnotation {
+
+  // Define the task queue name
+  static final String TASK_QUEUE = "CustomAnnotationTaskQueue";
+
+  // Define our workflow unique id
+  static final String WORKFLOW_ID = "CustomAnnotationWorkflow";
+
+  /**
+   * The Workflow Definition's Interface must contain one method annotated with @WorkflowMethod.
+   *
+   * <p>Workflow Definitions should not contain any heavyweight computations, non-deterministic
+   * code, network calls, database operations, etc. Those things should be handled by the
+   * Activities.
+   *
+   * @see WorkflowInterface
+   * @see WorkflowMethod
+   */
+  @WorkflowInterface
+  public interface GreetingWorkflow {
+
+    /**
+     * This is the method that is executed when the Workflow Execution is started. The Workflow
+     * Execution completes when this method finishes execution.
+     */
+    @WorkflowMethod
+    String getGreeting(String name);
+  }
+
+  /**
+   * This is the Activity Definition's Interface. Activities are building blocks of any Temporal
+   * Workflow and contain any business logic that could perform long running computation, network
+   * calls, etc.
+   *
+   * <p>Annotating Activity Definition methods with @ActivityMethod is optional.
+   *
+   * @see ActivityInterface
+   * @see io.temporal.activity.ActivityMethod
+   */
+  @ActivityInterface
+  public interface GreetingActivities {
+
+    /** Define your activity method which can be called during workflow execution */
+    String composeGreeting(String greeting, String name);
+  }
+
+  // Define the workflow implementation which implements our getGreeting workflow method.
+  public static class GreetingWorkflowImpl implements GreetingWorkflow {
+
+    /**
+     * Define the GreetingActivities stub. Activity stubs are proxies for activity invocations that
+     * are executed outside of the workflow thread on the activity worker, that can be on a
+     * different host. Temporal is going to dispatch the activity results back to the workflow and
+     * unblock the stub as soon as activity is completed on the activity worker.
+     *
+     * <p>In the {@link ActivityOptions} definition the "setStartToCloseTimeout" option sets the
+     * maximum time of a single Activity execution attempt. For this example it is set to 10
+     * seconds.
+     *
+     * <p>In the {@link ActivityOptions} definition the "setInitialInterval" option sets the
+     * interval of the first retry. It is set to 1 second. The "setDoNotRetry" option is a list of
+     * application failures for which retries should not be performed.
+     *
+     * <p>By default the maximum number of retry attempts is set to "unlimited" however you can
+     * change it by adding the "setMaximumAttempts" option to the retry options.
+     */
+    private final GreetingActivities activities =
+        Workflow.newActivityStub(
+            GreetingActivities.class,
+            ActivityOptions.newBuilder()
+                .setStartToCloseTimeout(Duration.ofSeconds(10))
+                .setRetryOptions(
+                    RetryOptions.newBuilder()
+                        .setInitialInterval(Duration.ofSeconds(1))
+                        .setDoNotRetry(IllegalArgumentException.class.getName())
+                        .build())
+                .build());
+
+    @Override
+    public String getGreeting(String name) {
+      // This is a blocking call that returns only after activity is completed.
+      return activities.composeGreeting("Hello", name);
+    }
+  }
+
+  /**
+   * Implementation of our workflow activity interface. It overwrites our defined composeGreeting
+   * activity method.
+   */
+  static class GreetingActivitiesImpl implements GreetingActivities {
+    private int callCount;
+    private long lastInvocationTime;
+
+    /**
+     * Our activity implementation simulates a failure 3 times. Given our previously set
+     * RetryOptions, our workflow is going to retry our activity execution.
+     */
+    @Override
+    @NextRetryDelay(failureType = "java.lang.IllegalStateException", delaySeconds = 2)
+    public synchronized String composeGreeting(String greeting, String name) {
+      if (lastInvocationTime != 0) {
+        long timeSinceLastInvocation = System.currentTimeMillis() - lastInvocationTime;
+        System.out.print(timeSinceLastInvocation + " milliseconds since last invocation. ");
+      }
+      lastInvocationTime = System.currentTimeMillis();
+      if (++callCount < 4) {
+        System.out.println("composeGreeting activity is going to fail");
+
+        /*
+         * We throw IllegalStateException here. It is not in the list of "do not retry" exceptions
+         * set in our RetryOptions, so a workflow retry is going to be issued
+         */
+        throw new IllegalStateException("not yet");
+      }
+
+      // after 3 unsuccessful retries we finally can complete our activity execution
+      System.out.println("composeGreeting activity is going to complete");
+      return greeting + " " + name + "!";
+    }
+  }
+
+  /**
+   * With our Workflow and Activities defined, we can now start execution. The main method starts
+   * the worker and then the workflow.
+   */
+  public static void main(String[] args) {
+
+    // Get a Workflow service stub.
+    WorkflowServiceStubs service = WorkflowServiceStubs.newLocalServiceStubs();
+
+    /*
+     * Get a Workflow service client which can be used to start, Signal, and Query Workflow Executions.
+     */
+    WorkflowClient client = WorkflowClient.newInstance(service);
+
+    /*
+     * Define the workflow factory. It is used to create workflow workers for a specific task queue.
+     */
+    WorkerFactory factory =
+        WorkerFactory.newInstance(
+            client,
+            WorkerFactoryOptions.newBuilder()
+                .setWorkerInterceptors(new NextRetryDelayActivityAnnotationInterceptor())
+                .build());
+
+    /*
+     * Define the workflow worker. Workflow workers listen to a defined task queue and process
+     * workflows and activities.
+     */
+    Worker worker = factory.newWorker(TASK_QUEUE);
+
+    /*
+     * Register our workflow implementation with the worker.
+     * Workflow implementations must be known to the worker at runtime in
+     * order to dispatch workflow tasks.
+     */
+    worker.registerWorkflowImplementationTypes(GreetingWorkflowImpl.class);
+
+    /*
+     * Register our Activity Types with the Worker. Since Activities are stateless and thread-safe,
+     * the Activity Type is a shared instance.
+     */
+    worker.registerActivitiesImplementations(new GreetingActivitiesImpl());
+
+    /*
+     * Start all the workers registered for a specific task queue.
+     * The started workers then start polling for workflows and activities.
+     */
+    factory.start();
+
+    // Set our workflow options
+    WorkflowOptions workflowOptions =
+        WorkflowOptions.newBuilder().setWorkflowId(WORKFLOW_ID).setTaskQueue(TASK_QUEUE).build();
+
+    // Create the workflow client stub. It is used to start our workflow execution.
+    GreetingWorkflow workflow = client.newWorkflowStub(GreetingWorkflow.class, workflowOptions);
+
+    /*
+     * Execute our workflow and wait for it to complete. The call to our getGreeting method is
+     * synchronous.
+     *
+     * See {@link io.temporal.samples.hello.HelloSignal} for an example of starting workflow
+     * without waiting synchronously for its result.
+     */
+    String greeting = workflow.getGreeting("World");
+
+    // Display workflow execution results
+    System.out.println(greeting);
+    System.exit(0);
+  }
+}

core/src/main/java/io/temporal/samples/customannotation/NextRetryDelay.java

@@ -0,0 +1,30 @@
+package io.temporal.samples.customannotation;
+
+import java.lang.annotation.*;
+
+/**
+ * NextRetryDelay is an annotation that can be used to specify the next retry delay for a particular
+ * failure type in a Temporal activity. It is used to provide a custom fixed delay if the activity
+ * fails with a specific exception type.
+ *
+ * <p>For this annotation to work, {@link NextRetryDelayActivityAnnotationInterceptor} must be
+ * passed as a worker interceptor to the worker factory.
+ */
+@Documented
+@Target(ElementType.METHOD)
+@Repeatable(NextRetryDelays.class)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface NextRetryDelay {
+  /**
+   * failureType is the type of failure that this retry delay applies to. It should be the fully
+   * qualified class name of the exception type or the type of the {@link
+   * io.temporal.failure.ApplicationFailure}.
+   */
+  String failureType();
+
+  /**
+   * delaySeconds is the fixed delay in seconds that should be applied for the specified failure
+   * type.
+   */
+  int delaySeconds();
+}

core/src/main/java/io/temporal/samples/customannotation/NextRetryDelayActivityAnnotationInterceptor.java

@@ -0,0 +1,115 @@
+/*
+ *  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.customannotation;
+
+import io.temporal.activity.ActivityExecutionContext;
+import io.temporal.common.interceptors.ActivityInboundCallsInterceptor;
+import io.temporal.common.interceptors.WorkerInterceptorBase;
+import io.temporal.common.metadata.POJOActivityImplMetadata;
+import io.temporal.common.metadata.POJOActivityMethodMetadata;
+import io.temporal.failure.ApplicationFailure;
+import io.temporal.failure.TemporalFailure;
+import java.lang.reflect.Method;
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Checks if the activity method has the @NextRetryDelay annotation. If it does, it will throw an
+ * ApplicationFailure with a delay set to the value of the annotation.
+ */
+public class NextRetryDelayActivityAnnotationInterceptor extends WorkerInterceptorBase {
+
+  @Override
+  public ActivityInboundCallsInterceptor interceptActivity(ActivityInboundCallsInterceptor next) {
+    return new ActivityInboundCallsInterceptorAnnotation(next);
+  }
+
+  public static class ActivityInboundCallsInterceptorAnnotation
+      extends io.temporal.common.interceptors.ActivityInboundCallsInterceptorBase {
+    private final ActivityInboundCallsInterceptor next;
+    private Map<String, Integer> delaysPerType = new HashMap<>();
+
+    public ActivityInboundCallsInterceptorAnnotation(ActivityInboundCallsInterceptor next) {
+      super(next);
+      this.next = next;
+    }
+
+    @Override
+    public void init(ActivityExecutionContext context) {
+      List<POJOActivityMethodMetadata> activityMethods =
+          POJOActivityImplMetadata.newInstance(context.getInstance().getClass())
+              .getActivityMethods();
+      // TODO: handle dynamic activity types
+      POJOActivityMethodMetadata currentActivityMethod =
+          activityMethods.stream()
+              .filter(x -> x.getActivityTypeName().equals(context.getInfo().getActivityType()))
+              .findFirst()
+              .get();
+      // Get the implementation method from the interface method
+      Method implementationMethod;
+      try {
+        implementationMethod =
+            context
+                .getInstance()
+                .getClass()
+                .getMethod(
+                    currentActivityMethod.getMethod().getName(),
+                    currentActivityMethod.getMethod().getParameterTypes());
+      } catch (NoSuchMethodException e) {
+        throw new RuntimeException(e);
+      }
+      // Get the @NextRetryDelay annotations from the implementation method
+      NextRetryDelay[] an = implementationMethod.getAnnotationsByType(NextRetryDelay.class);
+      for (NextRetryDelay a : an) {
+        delaysPerType.put(a.failureType(), a.delaySeconds());
+      }
+      next.init(context);
+    }
+
+    @Override
+    public ActivityOutput execute(ActivityInput input) {
+      if (delaysPerType.size() == 0) {
+        return next.execute(input);
+      }
+      try {
+        return next.execute(input);
+      } catch (ApplicationFailure ae) {
+        Integer delay = delaysPerType.get(ae.getType());
+        if (delay != null) {
+          // TODO: make sure to pass all the other parameters to the new ApplicationFailure
+          throw ApplicationFailure.newFailureWithCauseAndDelay(
+              ae.getMessage(), ae.getType(), ae.getCause(), Duration.ofSeconds(delay));
+        }
+        throw ae;
+      } catch (TemporalFailure tf) {
+        throw tf;
+      } catch (Exception e) {
+        Integer delay = delaysPerType.get(e.getClass().getName());
+        if (delay != null) {
+          throw ApplicationFailure.newFailureWithCauseAndDelay(
+              e.getMessage(), e.getClass().getName(), e, Duration.ofSeconds(delay));
+        }
+        throw e;
+      }
+    }
+  }
+}

core/src/main/java/io/temporal/samples/customannotation/NextRetryDelays.java

@@ -0,0 +1,11 @@
+package io.temporal.samples.customannotation;
+
+import java.lang.annotation.*;
+
+/** NextRetryDelays is a container annotation for multiple {@link NextRetryDelay} annotations. */
+@Documented
+@Target(ElementType.METHOD)
+@Retention(RetentionPolicy.RUNTIME)
+public @interface NextRetryDelays {
+  NextRetryDelay[] value();
+}

core/src/main/java/io/temporal/samples/customannotation/README.md

@@ -0,0 +1,7 @@
+# Custom annotation
+
+The sample demonstrates how to create a custom annotation using an interceptor. In this case the annotation allows specifying a fixed next retry delay for a certain failure type.
+
+```bash
+./gradlew -q execute -PmainClass=io.temporal.samples.customannotation.CustomAnnotation
+```