Async.function does not like Scala lambdas

[vitaliihonta]

Async.function does not like Scala lambdas

[ploddi]

I'm not 100% correct in prev comment: if you pass stub method to ZPromise it will be the same as in temporalio/sdk-java#1489

So ZPromise is safe for ordinary Scala lambdas, but for not for stubbed methods. For example you should not wrap child workflow invocations with ZPromise

[ricochet1k]

See also temporalio/sdk-java#1489

[vitaliihonta]

@ricochet1k Thank you for reporting! I'll also try to reproduce the problem

[ploddi]

After little research I believe that temporalio/sdk-java#1489 is misleading.

isAsync checks, if lambda passed to Async.function is a stub:

private static boolean isAsyncJava(Object func) {
    SerializedLambda lambda = JavaLambdaUtils.toSerializedLambda(func);
    Object target = JavaLambdaUtils.getTarget(lambda);
    return target instanceof ActivityStub
        || target instanceof ChildWorkflowStub
        || target instanceof ExternalWorkflowStub
        || (target instanceof AsyncInternal.AsyncMarker
            && lambda.getImplMethodKind() == MethodHandleInfo.REF_invokeInterface);
  }

specifically created by ActivityStub, ChildWorkflowStub, ExternalWorkflowStub or AsyncInternal.AsyncMarker itself.

In case of ZPromise I do not think that Scala lambda passed to Async.function expected to be async/stub in Temporal terms (because they are pure thunks). So seems like temporalio/sdk-java#1489 is not applicable to ZPromise implementation.

Explain why one should assume that isAsyncJava should return true for every Scala lambda, like () => Unit?

[scoplin]

I believe a pure thunk is also not detected as an async stub invocation. While it will run asynchronously, it won't necessarily do so in the way that we expect. The consequence was thus (as I noted in temporalio/sdk-java#1489):

If the Async.function (or similar) methods is called in an attempt to dispatch a child workflow from a Scala class, it fails to properly detect the attempt to call a workflow method on a temporal stub, and instead falls back to submitting the function to a new workflow thread in AsyncInternal. As a result, we saw excessive threads being created and consumed in workers, eventually resulting in RejectedExectionException being thrown from the underlying executor.

Specifically, it is this code in AsyncInternal#execute that has a problem:

    if (temporalStub) {
      initAsyncInvocation();
      try {
        func.apply();
        return getAsyncInvocationResult();
      } catch (Exception e) {
        return Workflow.newFailedPromise(Workflow.wrap(e));
      } finally {
        closeAsyncInvocation();
      }
    } else {
      CompletablePromise<R> result = Workflow.newPromise();
      WorkflowThread.newThread(
          () -> {
            try {
              result.complete(func.apply());
            } catch (Exception e) {
              result.completeExceptionally(Workflow.wrap(e));
            }
          },
          false);
      return result;
    }
  }

That second branch results in a situation that can draw the exception under too much concurrency.