Implement activity helpers

[pjgg]

As a developer I would like to be able to define or even implements
commons methods (default methods) that could be extended by an
activity method interface. This methods, should not be consider as an
activity method.

This feature will gives an upper level of abstraction allowing commons
behavior between activity methods.

In order to support this feature, an annotation named 'ActivityHelper'
was developed. You must add this annotation to all those interfaces that
gives some kind of support or gives an upper level of abstraction to an
activity method interface.

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.

---

pjgg seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[mfateev]

Sorry for the late reply, missed notification about the new PR.
Do I understand correctly that you want to have a base interface that two activities implement?

[mfateev]

I have an alternative proposal. Treat all base interfaces as non activity interfaces. Only the interfaces that an activity implementation directly implements are used to define activities. Example:

interface A {
   void foo();
}

interface B extends A {
   void bar();
}

interface C extends A {
}

class BImpl implements B {
  ...
}

class CImpl implements C {
 ..
}

worker.registerActivityImpl(new BImpl(), new CIml());

The resulting activities registered with the worker:
B::bar, B::foo, C::foo

[pjgg]

Thank you for your response @mfateev

The main problem that we have is not the hierarchy between interfaces. The problem is that this class POJOActivityTaskHandler.java (method: addActivityImplementation) will walk through all activity implementation interfaces, and will assume that all methods (event methods that are not annotated as an activity method) are activity methods.

So, If you have a look at the unit test that we have implemented

https://github.com/uber/cadence-java-client/pull/353/files#diff-95c5d7ffe18c3ca1e1816a77893fc38cR155

You will see that is pretty much the same as your example:

@ActivityHelper
  public interface TestParentActivity {
    String execute(String input);
  }

   public interface TestChild1Activity extends TestParentActivity {

     @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
    @Override
    String execute(String input);
  }

   public interface TestChild2Activity extends TestParentActivity {

     @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
    @Override
    String execute(String input);
  }

   private static class ActivityChild1Impl implements TestChild1Activity {

     @Override
    public String execute(String input) {
      return Activity.getTask().getActivityType() + "-" + input;
    }
  }

   private static class ActivityChild2Impl implements TestChild2Activity {

     @Override
    public String execute(String input) {
      return Activity.getTask().getActivityType() + "-" + input;
    }
  }

If you remove the annotation ActivityHelper and run the test, then you will get the following error:

java.lang.IllegalStateException: TestParentActivity::execute activity type is already registered with the worker
	at com.uber.cadence.internal.sync.POJOActivityTaskHandler.addActivityImplementation(POJOActivityTaskHandler.java:112)
	at com.uber.cadence.internal.sync.POJOActivityTaskHandler.setActivitiesImplementation(POJOActivityTaskHandler.java:164)
	at com.uber.cadence.internal.sync.SyncActivityWorker.setActivitiesImplementation(SyncActivityWorker.java:44)
	at com.uber.cadence.worker.Worker.registerActivitiesImplementations(Worker.java:280)
	at com.uber.cadence.internal.testing.WorkflowTestingTest.testChildActivity(WorkflowTestingTest.java:194)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:298)
	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:292)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.lang.Thread.run(Thread.java:748)

In order to avoid this behavior and allow the developer to have interfaces with methods (not activity methods) that could be extended by an interface that implements an activity method we have created an annotation:

https://github.com/uber/cadence-java-client/pull/353/files#diff-afc5b685d6cf32cbbdc56ae28d92888cR34

So all the interfaces that are annotated with this annotation will be ignored by POJOActivityTaskHandler.java (method: addActivityImplementation)

https://github.com/uber/cadence-java-client/pull/353/files#diff-c0d35c261ac27624c0c8ea09a069a4f3R101

[mfateev]

I see. How do you expect the activity stub to behave in this case?

interface A {
   void foo();
}

interface B extends A {
  @ActivityMethod
   void bar();
}

In the workflow code:

B b = Workflow.newActivityStub(B.class);
b.foo();

Should b.foo() fail at runtime?

[pjgg]

This example works, but this one doesn't

public interface A {
    String foo();
  }

  public interface B extends A {
    @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
    @Override
    String foo();
  }

  public interface C extends A {
    @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
    @Override
    String foo();
  }

You can test it running the following UnitTest:

public static class PabloActivityWorkflow implements TestPabloWorkflow {

    private final B activity1 = Workflow.newActivityStub(B.class);

    @Override
    public String workflow1() {
      return activity1.foo();
    }
  }

  public interface A {
    String foo();
  }

  public interface B extends A {
    @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
    @Override
    String foo();
  }

  public interface C extends A {
    @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
    @Override
    String foo();
  }

  @Test
  public void testChildA() {
    Worker worker = testEnvironment.newWorker(TASK_LIST);
    worker.registerWorkflowImplementationTypes(PabloActivityWorkflow.class);
    worker.registerActivitiesImplementations(new ActivityBImpl(), new ActivityCImpl());
    testEnvironment.start();
    WorkflowClient client = testEnvironment.newWorkflowClient();
    TestPabloWorkflow workflow = client.newWorkflowStub(TestPabloWorkflow.class);
    String result = workflow.workflow1();
    assertEquals(result, "eco");
  }

  private static class ActivityBImpl implements B {

    @Override
    public String foo() {
      return "eco";
    }
  }

  private static class ActivityCImpl implements C {

    @Override
    public String foo() {
      return "eco";
    }
  }

Because in the same worker, you have two activities with the same default name "A::foo"
So in order to reproduce the issue, you need at least two activities with the same activity-method name.

[mfateev]

I believe my initial proposal here (#353 (comment)) solves the issue in the last example.

[pjgg]

You need to have the same method name in all the interfaces in order to reproduce this issue. You could have a look the unit test that comes with this PR, or on the comments.

[mfateev]

Sorry, I'm still confused. Could you explain why exactly my proposal doesn't solve your problem?

[pjgg]

Because your proposal doesn't overwrite any method (is not the same scenario).

Please try this scenario in order to understand the issue. You will notice that cadence complain about 'foo' activity because is already registered.

interface A {
void foo();
}

interface B extends A {
void foo();
}

interface C extends A {
void foo();
}

class BImpl implements B {
...
}

class CImpl implements C {
..
}

worker.registerActivityImpl(new BImpl(), new CIml());

To avoid this weird behavior, we've created a new annotation '@ActivityHelper'. This PR talks about this scenario.

[mfateev]

Sorry, but my proposal solves your issue. It will register B::foo and C::foo, but is not going to register A::foo. What do I miss?

[pjgg]

Let me try to drive you, and see if together we can understand what is happening:

Imagine that you have the following activity hierarchy

public interface A {

    void foo();
  }

  public interface B extends A {

    @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
    @Override
    void foo();
  }

  public interface C extends A {

    @ActivityMethod(scheduleToCloseTimeoutSeconds = 3600)
    @Override
    void foo();
  }

  private static class Bimpl implements B {

    @Override
    public void foo() {
    }
  }

  private static class Cimpl implements C {

    @Override
    public void foo() {
    }
  }

And the following workflow (doesn't really care)

 public static class BandCActivityWorkflow implements TestWorkflow {

    private final B b = Workflow.newActivityStub(B.class);
    private final C c = Workflow.newActivityStub(C.class);

    @Override
    public String workflow1(String input) {
      Workflow.sleep(Duration.ofHours(1)); // test time skipping
      b.foo();
      c.foo();
      return "DONE";
    }
  }

When we try to run this workflow, by the following unit test:

  @Test
  public void testBandCActivity() {
    Worker worker = testEnvironment.newWorker(TASK_LIST);
    worker.registerWorkflowImplementationTypes(BandCActivityWorkflow.class);

    worker.registerActivitiesImplementations(new Bimpl(), new Cimpl());

    testEnvironment.start();
    WorkflowClient client = testEnvironment.newWorkflowClient();
    TestWorkflow workflow = client.newWorkflowStub(TestWorkflow.class);
    String result = workflow.workflow1("input1");
    assertEquals("DONE", result);
  }

We are getting the following exception:

java.lang.IllegalStateException: A::foo activity type is already registered with the worker
	at com.uber.cadence.internal.sync.POJOActivityTaskHandler.addActivityImplementation(POJOActivityTaskHandler.java:112)
	at com.uber.cadence.internal.sync.POJOActivityTaskHandler.setActivitiesImplementation(POJOActivityTaskHandler.java:164)
	at com.uber.cadence.internal.sync.SyncActivityWorker.setActivitiesImplementation(SyncActivityWorker.java:44)
	at com.uber.cadence.worker.Worker.registerActivitiesImplementations(Worker.java:297)
	at com.uber.cadence.internal.testing.WorkflowTestingTest.testBandCActivity(WorkflowTestingTest.java:195)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
	at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26)
	at org.junit.internal.runners.statements.RunAfters.evaluate(RunAfters.java:27)
	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:298)
	at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:292)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.lang.Thread.run(Thread.java:748)

Please copy this test, and run it by your self. Then you can debug cadence-java-client and pay attention to POJOActivityTaskHandler.java specially to addActivityImplementation method. This method will walk through all activity implementation interfaces and will assume that all methods (event those methods that are not annotated as an activity method) are activity methods. And this is why cadence-java-client is complaining about A::foo "activity".

java.lang.IllegalStateException: A::foo activity type is already registered with the worker

Because B and C activity has the same parent, and cadence will register this parent twice (when it should not been registered).

The solution that we thought was a new annotation called "activityHelper" in order to make cadence know, that this interface "A" should be not considered as an interface that holds activity methods.

You can also test this approach running the unitTest that comes with this PR.

[mfateev]

Here is my proposal (not yet implemented) that I put at the beginning of this thread:

Treat all base interfaces as non activity interfaces. Only the interfaces that an activity implementation directly implements are used to define activities. Example:

interface A {
   void foo();
}

interface B extends A {
   void bar();
}

interface C extends A {
}

class BImpl implements B {
  ...
}

class CImpl implements C {
 ..
}

worker.registerActivityImpl(new BImpl(), new CIml());

The resulting activities registered with the worker:
B::bar, B::foo, C::foo

Could you explain why exactly it is not solving your problem?

The issue with your proposal is the following:

How do you expect the activity stub to behave in this case?

interface A {
   void foo();
}

interface B extends A {
  @ActivityMethod
   void bar();
}

In the workflow code:

B b = Workflow.newActivityStub(B.class);
b.foo();

Should b.foo() fail at runtime as it is not annotated with @ActivityHelper?

[pjgg]

Could you explain why exactly it is not solving your problem?

Because you are not in the same Scenario that I told you. You only will be able to reproduce the issue if all the interfaces extend the same method name.

`
interface A {
void foo();
}

interface B extends A {
void foo();
}

interface C extends A {
void foo();
}

class BImpl implements B {
...
}

class CImpl implements C {
..
}

worker.registerActivityImpl(new BImpl(), new CIml());
`

I have tested all the examples that I gave you. And you can reproduce the issue with the unit test that comes with this PR.

You will get the following exception:

java.lang.IllegalStateException: A::foo activity type is already registered with the worker

Because as I told you:

B and C activity has the same parent, and cadence will register this parent twice (when it should not been registered).

You will find the problem in this class: POJOActivityTaskHandler.java -> addActivityImplementation

Look I am not going to continue with this issue. If you want to investigate it's up to you.
From my side, you could close the issue.

Thank you anyway!.