Implement asyncapi call + asyncapi DSL by mcruzdev · Pull Request #1617 · open-workflow-specification/sdk-java

mcruzdev · 2026-08-12T23:49:34Z

Many thanks for submitting your Pull Request ❤️!

What this PR does / why we need it:

Special notes for reviewers:

Additional information (if needed):

Copilot

Pull request overview

Adds AsyncAPI support to the Java workflow runtime by introducing a new impl/asyncapi module that implements the call: asyncapi task (publish + subscribe) and wires it into the runtime via SPI, along with test fixtures to validate behavior.

Changes:

Introduces impl/asyncapi module with AsyncAPIExecutor/builder, lightweight AsyncAPI document model + reader, and a pluggable AsyncApiChannelProvider SPI.
Adds JUnit tests plus AsyncAPI spec/workflow YAML fixtures for publish and multiple subscribe policies.
Updates Maven module wiring and test module dependencies to include the new asyncapi implementation.

Reviewed changes

Copilot reviewed 20 out of 20 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-until.yaml	Adds a workflow fixture for subscribe + `until` consumption policy.
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-foreach.yaml	Adds a workflow fixture for subscribe + `foreach` per-message processing.
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-filter.yaml	Adds a workflow fixture for subscribe + message filtering.
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-subscribe-amount.yaml	Adds a workflow fixture for subscribe + `amount` consumption policy.
impl/test/src/test/resources/workflows-samples/asyncapi/asyncapi-publish.yaml	Adds a workflow fixture for publish with payload + headers.
impl/test/src/test/resources/schema/asyncapi/asyncapi.yaml	Adds an AsyncAPI 3.0 test document fixture used by tests.
impl/test/src/test/java/io/serverlessworkflow/impl/test/AsyncAPITest.java	Adds integration-style tests using an in-memory provider and MockWebServer-hosted AsyncAPI spec.
impl/test/pom.xml	Adds test-scope dependency on `serverlessworkflow-impl-asyncapi`.
impl/pom.xml	Adds `serverlessworkflow-impl-asyncapi` to dependency management and module list.
impl/asyncapi/src/main/resources/META-INF/services/io.serverlessworkflow.impl.executors.CallableTaskBuilder	Registers `AsyncAPIExecutorBuilder` via Java SPI.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/UnifiedAsyncAPI.java	Adds a minimal unified AsyncAPI document model for server/channel/operation resolution.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiSubscriptionHandle.java	Adds subscription lifecycle abstraction (unsubscribe + closed future).
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIReader.java	Adds AsyncAPI document reader using workflow format mappers (YAML/JSON).
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiInboundMessage.java	Adds inbound message envelope (payload/headers/correlationId) with null-safe defaults.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutorBuilder.java	Builds executor configs (payload/header resolvers, predicates, foreach executor, timeout resolver).
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java	Implements publish/subscribe execution, document/server/channel resolution, and consumption policy evaluation.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelProvider.java	Adds transport SPI for publish and subscribe operations.
impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncApiChannelInfo.java	Adds resolved channel metadata passed to providers (URI, protocol, operation, auth token).
impl/asyncapi/pom.xml	Introduces the new `serverlessworkflow-impl-asyncapi` Maven module.
asyncapi-call-plan.md	Adds an implementation plan / design notes for the AsyncAPI call feature.

Suppressed comments (3)

impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java:246

AsyncApiSubscriptionHandle.closed() is never observed, so if the provider closes the subscription early (or fails) the task future may never complete (and can hang indefinitely unless a consume timeout is configured). Wire handle.closed() into the task result so provider-side shutdown/errors terminate the task deterministically.

    result.whenComplete((r, ex) -> handle.unsubscribe());

impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java:279

processTaskList(...).join() blocks the provider callback thread (and it’s currently executed while holding the synchronized (collection) lock). This can severely limit throughput and can deadlock/starve if the subscription callback is invoked on the same executor used to run workflow tasks. Prefer a non-blocking FIFO chain (e.g., keep a CompletableFuture<Void> tail and thenCompose per message) and avoid holding locks while waiting for workflow task completion.

    if (subscribeConfig.foreachExecutor() != null) {
      taskContext.variables().put(subscribeConfig.foreachItem(), messageModel);
      taskContext.variables().put(subscribeConfig.foreachAt(), collection.size());
      return TaskExecutorHelper.processTaskList(
              subscribeConfig.foreachExecutor(),
              workflowContext,
              Optional.of(taskContext),
              messageModel)
          .join();

impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutor.java:261

The executor implements consume.for timeout behavior (graceful completion with partial results), but there’s no test covering this path yet. Adding a test + workflow fixture that sets subscription.consume.for and verifies the result completes with a partial collection would protect this behavior from regressions.

    subscribeConfig
        .consumeTimeout()
        .ifPresent(
            resolver -> {
              Duration duration = resolver.apply(workflowContext, taskContext, input);
              CompletableFuture.delayedExecutor(duration.toMillis(), TimeUnit.MILLISECONDS)
                  .execute(
                      () -> {
                        synchronized (collection) {
                          if (!result.isDone()) {
                            result.complete(collection);
                          }
                        }
                      });
            });

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

+    AsyncAPIExecutor.PublishConfig publishConfig =
+        Optional.ofNullable(args.getMessage())
+            .map(msg -> buildPublishConfig(application, msg))
+            .orElse(null);
+
+    AsyncAPIExecutor.SubscribeConfig subscribeConfig =
+        Optional.ofNullable(args.getSubscription())
+            .map(sub -> buildSubscribeConfig(application, sub, position, definition))
+            .orElse(null);


+    AsyncApiMessageConsumptionPolicyUnion consumePolicy = subscription.getConsume();
+
+    Optional<WorkflowValueResolver<Duration>> consumeTimeout =
+        Optional.ofNullable(consumePolicy.get().getFor())
+            .map(t -> WorkflowUtils.fromTimeoutAfter(application, t));


+    if (channelName != null) {
+      return channelName;
+    }


+    <dependencies>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-impl-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.serverlessworkflow</groupId>
+            <artifactId>serverlessworkflow-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+    </dependencies>


Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>

Copilot

🟡 Changes recommended

The AsyncAPI executor/builder currently has validated correctness gaps (missing configuration validation and incomplete subscription lifecycle/error handling) that can lead to NPEs or hanging executions.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Review details

Suppressed comments (2)

impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutorBuilder.java:80

AsyncAPI call tasks can currently be configured with neither message nor subscription, or with both, which will lead to a NullPointerException (subscribeConfig == null) or ambiguous behavior (publish path always wins). Validate the configuration early and fail with a clear error.

    AsyncAPIExecutor.PublishConfig publishConfig =
        Optional.ofNullable(args.getMessage())
            .map(msg -> buildPublishConfig(application, msg))
            .orElse(null);

    AsyncAPIExecutor.SubscribeConfig subscribeConfig =
        Optional.ofNullable(args.getSubscription())
            .map(sub -> buildSubscribeConfig(application, sub, position, definition))
            .orElse(null);

impl/asyncapi/src/main/java/io/serverlessworkflow/impl/executors/asyncapi/AsyncAPIExecutorBuilder.java:122

subscription.getConsume() is used without a null-check; if a workflow provides subscription but omits consume, this will throw a NullPointerException during build. This should be validated with a clear error message (or defaulted).

    AsyncApiMessageConsumptionPolicyUnion consumePolicy = subscription.getConsume();

    Optional<WorkflowValueResolver<Duration>> consumeTimeout =
        Optional.ofNullable(consumePolicy.get().getFor())
            .map(t -> WorkflowUtils.fromTimeoutAfter(application, t));

Files reviewed: 42/42 changed files
Comments generated: 6
Review effort level: Lite

+    result.whenComplete((r, ex) -> handle.unsubscribe());
+


+            msg -> {
+              synchronized (collection) {
+                if (result.isDone()) {
+                  return;
+                }
+                WorkflowModel messageModel = toWorkflowModel(factory, msg);
+                if (subscribeConfig.filterPredicate().isPresent()
+                    && !subscribeConfig
+                        .filterPredicate()
+                        .get()
+                        .test(workflowContext, taskContext, messageModel)) {
+                  return;
+                }
+                WorkflowModel processedModel =
+                    processMessage(messageModel, collection, workflowContext, taskContext);
+                collection.add(processedModel);
+                if (isConsumptionPolicySatisfied(workflowContext, taskContext, collection)) {
+                  result.complete(collection);
+                }
+              }
+            });


+**Modules touched**: `impl/asyncapi` only. No changes to `impl/core`, `types`, `api`,
+or any other module.


+                new IllegalStateException(
+                    "Missing AsyncApiChannelProvider. Register one via"
+                        + " WorkflowApplication.builder().withAdditionalObject(\""
+                        + AsyncApiChannelProvider.ASYNC_API_CHANNEL_PROVIDER
+                        + "\", provider)"));


  public static final String HTTP_REQUEST_DECORATOR_KEY = "HttpRequestDecorators";
  private final WorkflowDefinition definition;
-  private final List<HttpRequestDecorator> requestDecorators;
+  private List<HttpRequestDecorator> requestDecorators;
  private WorkflowValueResolver<URI> pathSupplier;
  private Object body;


+    try (InputStream is =
+        AsyncAPITest.class.getResourceAsStream("/schema/asyncapi/asyncapi.yaml")) {
+      asyncApiSpec = is.readAllBytes();
+    }


mcruzdev requested a review from fjtirado as a code owner August 12, 2026 23:49

Copilot AI lite review requested due to automatic review settings August 12, 2026 23:49

Copilot started reviewing on behalf of mcruzdev August 12, 2026 23:50 View session

Copilot AI reviewed Aug 12, 2026

View reviewed changes

mcruzdev marked this pull request as draft August 12, 2026 23:58

mcruzdev changed the title ~~Implement asyncapi call~~ Implement asyncapi call + asyncapi DSL Aug 13, 2026

Add initial asyncapi implementation + asyncapi DSL

dc153e7

Signed-off-by: Matheus Cruz <matheuscruz.dev@gmail.com>

mcruzdev force-pushed the issue-1607 branch from 98574db to dc153e7 Compare August 13, 2026 01:29

mcruzdev removed the request for review from fjtirado September 9, 2026 03:11

mcruzdev assigned fjtirado and unassigned fjtirado Sep 9, 2026

mcruzdev marked this pull request as ready for review September 9, 2026 03:12

Copilot AI review requested due to automatic review settings September 9, 2026 03:12

mcruzdev marked this pull request as draft September 9, 2026 03:12

Copilot started reviewing on behalf of mcruzdev September 9, 2026 03:12 View session

Copilot AI reviewed Sep 9, 2026

View reviewed changes

Navigation Menu

Uh oh!

Implement asyncapi call + asyncapi DSL - #1617

Implement asyncapi call + asyncapi DSL#1617
mcruzdev wants to merge 1 commit into
open-workflow-specification:mainfrom
mcruzdev:issue-1607

mcruzdev commented Aug 12, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		Modules touched: `impl/asyncapi` only. No changes to `impl/core`, `types`, `api`,
		or any other module.

Uh oh!

Conversation

mcruzdev commented Aug 12, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

🟡 Changes recommended

Suppressed comments (2)

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

mcruzdev commented Aug 12, 2026 •

edited

Loading