version-test

gistfile1.txt

/*
 *  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.hello;

import io.nexusrpc.handler.OperationContext;
import io.temporal.activity.ActivityInterface;
import io.temporal.activity.ActivityOptions;
import io.temporal.client.WorkflowOptions;
import io.temporal.common.interceptors.*;
import io.temporal.testing.TestEnvironmentOptions;
import io.temporal.testing.TestWorkflowEnvironment;
import io.temporal.testing.TestWorkflowRule;
import io.temporal.testing.WorkflowReplayer;
import io.temporal.worker.Worker;
import io.temporal.worker.WorkerFactoryOptions;
import io.temporal.workflow.Workflow;
import io.temporal.workflow.WorkflowInterface;
import io.temporal.workflow.WorkflowMethod;
import java.time.Duration;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;

public class VersioningInterceptorTest {
  @Rule
  public TestWorkflowRule testWorkflowRule =
      TestWorkflowRule.newBuilder().setDoNotStart(true).build();

  @Test
  public void TestReplayFromVersionOneToTwo() {
    TestWorkflowEnvironment testEnvDefaultVersion =
        TestWorkflowEnvironment.newInstance(
            TestEnvironmentOptions.newBuilder()
                .setWorkerFactoryOptions(
                    WorkerFactoryOptions.newBuilder()
                        // set use default version
                        .setWorkerInterceptors(new TestWorkerInterceptor(true))
                        .build())
                .build());

    Worker defaultVersionWorker = testEnvDefaultVersion.newWorker("test-tq");
    defaultVersionWorker.registerWorkflowImplementationTypes(TestWorkflowImpl.class);
    defaultVersionWorker.registerActivitiesImplementations(new TestActivitiesImpl());

    TestWorkflowEnvironment testEnvV1 =
        TestWorkflowEnvironment.newInstance(
            TestEnvironmentOptions.newBuilder()
                .setWorkerFactoryOptions(
                    WorkerFactoryOptions.newBuilder()
                        // set not to use default version
                        .setWorkerInterceptors(new TestWorkerInterceptor(false))
                        .build())
                .build());

    Worker v1Worker = testEnvV1.newWorker("test-tq");
    v1Worker.registerWorkflowImplementationTypes(TestWorkflowImpl.class);
    v1Worker.registerActivitiesImplementations(new TestActivitiesImpl());

    // run default version wf
    testEnvDefaultVersion.start();
    TestWorkflow workflow =
        testEnvDefaultVersion
            .getWorkflowClient()
            .newWorkflowStub(
                TestWorkflow.class,
                WorkflowOptions.newBuilder()
                    .setTaskQueue("test-tq")
                    .setWorkflowId("test-wf")
                    .build());
    workflow.exec("test input");

    // Get this runs exec history
    String defaultVersionHistory =
        testEnvDefaultVersion.getWorkflowClient().fetchHistory("test-wf").toJson(true);
    testEnvDefaultVersion.shutdown();

    // run replayer with worker from testEnvV1
    testEnvV1.start();
    try {
      WorkflowReplayer.replayWorkflowExecution(
          io.temporal.internal.common.WorkflowExecutionHistory.fromJson(defaultVersionHistory),
          v1Worker);
    } catch (Exception e) {
      Assert.fail(e.getMessage());
    }
    testEnvV1.shutdown();
  }

  @Test(expected = Exception.class)
  public void TestReplayForMissingActivity() throws Exception {
    TestWorkflowEnvironment testEnvDefaultVersion =
        TestWorkflowEnvironment.newInstance(
            TestEnvironmentOptions.newBuilder()
                .setWorkerFactoryOptions(
                    WorkerFactoryOptions.newBuilder()
                        // set use default version
                        .setWorkerInterceptors(new TestWorkerInterceptor(true))
                        .build())
                .build());

    // Create two workers, one to get event history and we will use the other for replayer
    // Each has diff workflow type registered to test missing activity

    Worker worker3Activity = testEnvDefaultVersion.newWorker("test-tq");
    worker3Activity.registerWorkflowImplementationTypes(TestWorkflowImpl.class);
    worker3Activity.registerActivitiesImplementations(new TestActivitiesImpl());

    Worker worker2Activity = testEnvDefaultVersion.newWorker("test-tq-replayer");
    worker2Activity.registerWorkflowImplementationTypes(TestWorkflowImplTwo.class);
    worker2Activity.registerActivitiesImplementations(new TestActivitiesImpl());

    testEnvDefaultVersion.start();

    TestWorkflow workflow =
        testEnvDefaultVersion
            .getWorkflowClient()
            .newWorkflowStub(
                TestWorkflow.class,
                WorkflowOptions.newBuilder()
                    .setTaskQueue("test-tq")
                    .setWorkflowId("test-wf")
                    .build());
    workflow.exec("test input");
    String defaultVersionHistory =
        testEnvDefaultVersion.getWorkflowClient().fetchHistory("test-wf").toJson(true);

    try {
      WorkflowReplayer.replayWorkflowExecution(
          io.temporal.internal.common.WorkflowExecutionHistory.fromJson(defaultVersionHistory),
          worker2Activity);
    } catch (Exception e) {
      throw e;
    }

    testEnvDefaultVersion.shutdown();
  }

  // Test Worker and Workflow Interceptors
  public static class TestWorkerInterceptor implements WorkerInterceptor {
    private boolean useDefaultVersion = false;

    public TestWorkerInterceptor(boolean useDefaultVersion) {
      this.useDefaultVersion = useDefaultVersion;
    }

    @Override
    public WorkflowInboundCallsInterceptor interceptWorkflow(WorkflowInboundCallsInterceptor next) {
      return new TestWorkflowInboundCallsInterceptor(useDefaultVersion, next);
    }

    @Override
    public ActivityInboundCallsInterceptor interceptActivity(ActivityInboundCallsInterceptor next) {
      return next;
    }

    @Override
    public NexusOperationInboundCallsInterceptor interceptNexusOperation(
        OperationContext context, NexusOperationInboundCallsInterceptor next) {
      return next;
    }
  }

  public static class TestWorkflowInboundCallsInterceptor
      extends WorkflowInboundCallsInterceptorBase {
    private boolean useDefaultVersion;

    public TestWorkflowInboundCallsInterceptor(
        Boolean useDefaultVersion, WorkflowInboundCallsInterceptor next) {
      super(next);
      this.useDefaultVersion = useDefaultVersion;
    }

    @Override
    public void init(WorkflowOutboundCallsInterceptor outboundCalls) {
      super.init(new TestWorkflowOutboundCallsInterceptor(useDefaultVersion, outboundCalls));
    }
  }

  public static class TestWorkflowOutboundCallsInterceptor
      extends WorkflowOutboundCallsInterceptorBase {
    private boolean useDefaultVersion = false;

    public TestWorkflowOutboundCallsInterceptor(
        boolean useDefaultVersion, WorkflowOutboundCallsInterceptor next) {
      super(next);
      this.useDefaultVersion = useDefaultVersion;
    }

    @Override
    public int getVersion(String changeId, int minSupported, int maxSupported) {
      if (useDefaultVersion) {
        return super.getVersion(changeId, Workflow.DEFAULT_VERSION, Workflow.DEFAULT_VERSION);
      } else {
        return super.getVersion(changeId, minSupported, maxSupported);
      }
    }
  }

  // Test Workflows and Activities
  @WorkflowInterface
  public interface TestWorkflow {
    @WorkflowMethod
    String exec(String input);
  }

  @WorkflowInterface
  public interface TestWorkflowTwo {
    @WorkflowMethod(name = "TestWorkflow")
    String exec(String input);
  }

  @ActivityInterface
  public interface TestActivities {
    String step1(String input);

    String step2(String input);

    String step3(String input);

    String step1New(String input);

    String step2New(String input);

    String step3New(String input);
  }

  public static class TestWorkflowImpl implements TestWorkflow {
    TestActivities activities =
        Workflow.newActivityStub(
            TestActivities.class,
            ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(2)).build());

    @Override
    public String exec(String input) {
      int version = Workflow.getVersion("change", Workflow.DEFAULT_VERSION, 1);

      if (version == Workflow.DEFAULT_VERSION) {
        activities.step1(input);
        activities.step2(input);
        activities.step3(input);
      } else {
        activities.step1New(input);
        activities.step2New(input);
        activities.step3New(input);
      }
      return "done";
    }
  }

  public static class TestWorkflowImplTwo implements TestWorkflowTwo {
    TestActivities activities =
        Workflow.newActivityStub(
            TestActivities.class,
            ActivityOptions.newBuilder().setStartToCloseTimeout(Duration.ofSeconds(2)).build());

    @Override
    public String exec(String input) {
      int version = Workflow.getVersion("change", Workflow.DEFAULT_VERSION, 1);

      if (version == Workflow.DEFAULT_VERSION) {
        activities.step1(input);
        activities.step2(input);
        //                activities.step3(input);
      } else {
        activities.step1New(input);
        activities.step2New(input);
        activities.step3New(input);
      }
      return "done";
    }
  }

  public static class TestActivitiesImpl implements TestActivities {
    @Override
    public String step1(String input) {
      return input;
    }

    @Override
    public String step2(String input) {
      return input;
    }

    @Override
    public String step3(String input) {
      return input;
    }

    @Override
    public String step1New(String input) {
      return input;
    }

    @Override
    public String step2New(String input) {
      return input;
    }

    @Override
    public String step3New(String input) {
      return input;
    }
  }
}