Throw clear error if workflow stub is misused

Author: Quinn-With-Two-Ns

temporal-sdk/src/main/java/io/temporal/internal/sync/ActivityStubImpl.java

@@ -13,6 +13,13 @@ final class ActivityStubImpl extends ActivityStubBase {
   private final WorkflowOutboundCallsInterceptor activityExecutor;
   private final Functions.Proc assertReadOnly;
 
+  private void assertSameWorkflow() {
+    if (activityExecutor != WorkflowInternal.getWorkflowOutboundInterceptor()) {
+      throw new IllegalStateException(
+          "Activity stub belongs to a different workflow. Create a new stub for each workflow instance.");
+    }
+  }
+
   static ActivityStub newInstance(
       ActivityOptions options,
       WorkflowOutboundCallsInterceptor activityExecutor,
@@ -34,6 +41,7 @@ static ActivityStub newInstance(
   @Override
   public <R> Promise<R> executeAsync(
       String activityName, Class<R> resultClass, Type resultType, Object... args) {
+    assertSameWorkflow();
     this.assertReadOnly.apply();
     return activityExecutor
         .executeActivity(

temporal-sdk/src/main/java/io/temporal/internal/sync/ChildWorkflowStubImpl.java

@@ -18,6 +18,13 @@ class ChildWorkflowStubImpl implements ChildWorkflowStub {
   private final CompletablePromise<WorkflowExecution> execution;
   private final Functions.Proc1<String> assertReadOnly;
 
+  private void assertSameWorkflow() {
+    if (outboundCallsInterceptor != WorkflowInternal.getWorkflowOutboundInterceptor()) {
+      throw new IllegalStateException(
+          "Child workflow stub belongs to a different workflow. Create a new stub for each workflow instance.");
+    }
+  }
+
   ChildWorkflowStubImpl(
       String workflowType,
       ChildWorkflowOptions options,
@@ -60,6 +67,7 @@ public <R> R execute(Class<R> resultClass, Object... args) {
 
   @Override
   public <R> R execute(Class<R> resultClass, Type resultType, Object... args) {
+    assertSameWorkflow();
     assertReadOnly.apply("schedule child workflow");
     Promise<R> result = executeAsync(resultClass, resultType, args);
     if (AsyncInternal.isAsync()) {
@@ -83,6 +91,7 @@ public <R> Promise<R> executeAsync(Class<R> resultClass, Object... args) {
 
   @Override
   public <R> Promise<R> executeAsync(Class<R> resultClass, Type resultType, Object... args) {
+    assertSameWorkflow();
     assertReadOnly.apply("schedule child workflow");
     ChildWorkflowOutput<R> result =
         outboundCallsInterceptor.executeChildWorkflow(
@@ -100,6 +109,7 @@ public <R> Promise<R> executeAsync(Class<R> resultClass, Type resultType, Object
 
   @Override
   public void signal(String signalName, Object... args) {
+    assertSameWorkflow();
     assertReadOnly.apply("signal workflow");
     Promise<Void> signaled =
         outboundCallsInterceptor

temporal-sdk/src/main/java/io/temporal/internal/sync/ExternalWorkflowStubImpl.java

@@ -13,6 +13,13 @@ class ExternalWorkflowStubImpl implements ExternalWorkflowStub {
   private final WorkflowExecution execution;
   private Functions.Proc1<String> assertReadOnly;
 
+  private void assertSameWorkflow() {
+    if (outboundCallsInterceptor != WorkflowInternal.getWorkflowOutboundInterceptor()) {
+      throw new IllegalStateException(
+          "External workflow stub belongs to a different workflow. Create a new stub for each workflow instance.");
+    }
+  }
+
   public ExternalWorkflowStubImpl(
       WorkflowExecution execution,
       WorkflowOutboundCallsInterceptor outboundCallsInterceptor,
@@ -29,6 +36,7 @@ public WorkflowExecution getExecution() {
 
   @Override
   public void signal(String signalName, Object... args) {
+    assertSameWorkflow();
     assertReadOnly.apply("signal external workflow");
     Promise<Void> signaled =
         outboundCallsInterceptor
@@ -52,6 +60,7 @@ public void signal(String signalName, Object... args) {
 
   @Override
   public void cancel() {
+    assertSameWorkflow();
     assertReadOnly.apply("cancel external workflow");
     Promise<Void> cancelRequested =
         outboundCallsInterceptor

temporal-sdk/src/main/java/io/temporal/internal/sync/LocalActivityStubImpl.java

@@ -13,6 +13,13 @@ class LocalActivityStubImpl extends ActivityStubBase {
   private final WorkflowOutboundCallsInterceptor activityExecutor;
   private final Functions.Proc assertReadOnly;
 
+  private void assertSameWorkflow() {
+    if (activityExecutor != WorkflowInternal.getWorkflowOutboundInterceptor()) {
+      throw new IllegalStateException(
+          "Local Activity stub belongs to a different workflow. Create a new stub for each workflow instance.");
+    }
+  }
+
   static ActivityStub newInstance(
       LocalActivityOptions options,
       WorkflowOutboundCallsInterceptor activityExecutor,
@@ -34,6 +41,7 @@ private LocalActivityStubImpl(
   @Override
   public <R> Promise<R> executeAsync(
       String activityName, Class<R> resultClass, Type resultType, Object... args) {
+    assertSameWorkflow();
     this.assertReadOnly.apply();
     return activityExecutor
         .executeLocalActivity(

temporal-sdk/src/main/java/io/temporal/internal/sync/NexusServiceStubImpl.java

@@ -13,6 +13,13 @@ public class NexusServiceStubImpl implements NexusServiceStub {
   private final WorkflowOutboundCallsInterceptor outboundCallsInterceptor;
   private final Functions.Proc1<String> assertReadOnly;
 
+  private void assertSameWorkflow() {
+    if (outboundCallsInterceptor != WorkflowInternal.getWorkflowOutboundInterceptor()) {
+      throw new IllegalStateException(
+          "Nexus service stub belongs to a different workflow. Create a new stub for each workflow instance.");
+    }
+  }
+
   public NexusServiceStubImpl(
       String name,
       NexusServiceOptions options,
@@ -31,6 +38,7 @@ public <R> R execute(String operationName, Class<R> resultClass, Object arg) {
 
   @Override
   public <R> R execute(String operationName, Class<R> resultClass, Type resultType, Object arg) {
+    assertSameWorkflow();
     assertReadOnly.apply("execute nexus operation");
     Promise<R> result = executeAsync(operationName, resultClass, resultType, arg);
     if (AsyncInternal.isAsync()) {
@@ -55,6 +63,7 @@ public <R> Promise<R> executeAsync(String operationName, Class<R> resultClass, O
   @Override
   public <R> Promise<R> executeAsync(
       String operationName, Class<R> resultClass, Type resultType, Object arg) {
+    assertSameWorkflow();
     assertReadOnly.apply("execute nexus operation");
     NexusOperationOptions mergedOptions =
         NexusOperationOptions.newBuilder(options.getOperationOptions())
@@ -82,6 +91,7 @@ public <R> NexusOperationHandle<R> start(String operationName, Class<R> resultCl
   @Override
   public <R> NexusOperationHandle<R> start(
       String operationName, Class<R> resultClass, Type resultType, Object arg) {
+    assertSameWorkflow();
     assertReadOnly.apply("schedule nexus operation");
     NexusOperationOptions mergedOptions =
         NexusOperationOptions.newBuilder(options.getOperationOptions())