Await Runtime Setup

Await steps model external boundaries inside QUEUE_ASYNC execution. TPF persists the interaction, dispatches through the configured adapter, and admits correlated completions back into the owning execution. Scalar and recovery paths suspend as WAITING_EXTERNAL; brokered itemized streams can also flow through a live await session while the transition is active.

For modeling guidance, start with Await Boundaries. For production operation, see Await Boundary Operations. Internally, await is backed by durable await units; for implementation diagrams and the state model, see Await Unit Runtime.

Supported Runtime Shapes

Cardinality	Interaction unit	Replay shape	App guidance
ONE_TO_ONE	one input unit, one external interaction	one output unit	Use for human approval, webhook callback, or brokered request/reply that returns one result.
ONE_TO_ONE over a stream	one owning unit with one item interaction per input item	completed item outputs replayed in input order	Use when each stream item has its own external decision.
ONE_TO_MANY	one input unit, one external interaction	one materialized multi-item output unit replayed as a stream	Keep completion payloads bounded.
MANY_TO_ONE	one materialized input unit, one external interaction	one output unit	Use when the external system decides on the whole batch.
MANY_TO_MANY	one materialized input unit, one external interaction	one materialized multi-item output unit replayed as a stream	Keep input and completion payloads bounded.

csv-payments uses authored ONE_TO_ONE await over a stream of PaymentRecord items. That is a stream of unary await interactions, not a hidden dispatch mode.

Itemized Queue-Async Mechanics

When ONE_TO_ONE await receives a stream, TPF creates one owning await unit and one interaction per input item. The unit gives the whole boundary one durable identity, while each item keeps its own correlation id, request payload, response payload, and item index.

For brokered await transports such as Kafka, the normal path is a live await session:

1. the source stream dispatches item interactions up to the configured live in-flight window,
2. each provider completion is recorded against its interaction before it is emitted,
3. the live session emits completed items to the resumed segment only as downstream requests them,
4. the source parser receives more demand as accepted completions free capacity.

This is still durable await, not a plain in-memory request/reply stream. If the worker crashes, the live session is cancelled, or a completion arrives with no live session, the queue-async coordinator falls back to durable item continuation:

1. the source stream must finish dispatching the unit and persist dispatchComplete,
2. the parent execution must be durably parked as WAITING_EXTERNAL for the same await unit,
3. the item completion must be recorded against the expected interaction,
4. only then can the queue-async coordinator dispatch that item's continuation.

This handles crash recovery, fast providers, and broker redelivery safely. A completion that cannot be accepted by a live session is recorded, then released through durable continuation only when the parent execution is actually waiting on that unit. Duplicate completions resolve through the same interaction record instead of re-running the continuation.

For csv-payments, Process Csv Payments Input emits PaymentRecord rows incrementally, Await Payment Provider dispatches each row as an item interaction, and Process Payment Status runs as completions are accepted by the live session or durable fallback. Terminal Object Publish writes PaymentOutput objects before the execution is marked successful.

The built-in interaction-api adapter is for human/UI inboxes and mock-provider style flows where another client queries pending interactions and later calls the generated completion API. The built-in webhook adapter dispatches an HTTP request to an external system and includes a signed resume token in the envelope. The built-in kafka adapter publishes a request envelope to Kafka and admits completion envelopes from a configured response channel. The built-in sqs adapter does the same request/completion pattern with SQS standard queues.

For runnable examples, use examples/restaurant-approval for human/UI await and examples/csv-payments for brokered unary await over a stream.

Runtime Guardrails

Await has the same side-effect rule as the rest of QUEUE_ASYNC: orchestrator state transitions are guarded, but external dispatch and external side effects are at-least-once. Use stable business idempotency keys at the external boundary.

Aggregate await shapes materialize input and/or output units in the current runtime. Do not use unbounded payloads for ONE_TO_MANY, MANY_TO_ONE, or MANY_TO_MANY await boundaries. If replay of a materialized multi-item output fails halfway through downstream execution, TPF restarts that output unit as a whole; it does not claim exactly-once partial stream progress inside the unit.

The runtime also enforces aggregate materialization guardrails:

Config key	Default	Applies to
pipeline.orchestrator.await-aggregate-max-input-items	10000	materialized input units for MANY_TO_ONE and MANY_TO_MANY await steps
pipeline.orchestrator.await-aggregate-max-output-items	10000	materialized output units for ONE_TO_MANY and MANY_TO_MANY await steps

Set either value to 0 only when the application has its own upstream size control and storage budget. Prefer stable business limits at the API/file/broker boundary rather than relying on these guards as the first line of defense.

Transport choice changes operational responsibility. interaction-api requires an API consumer to query and complete pending work. webhook requires stable resume-token signing and callback reachability. kafka requires broker channel configuration, consumer health, and response-envelope monitoring. sqs requires request/response queue configuration, poller health, visibility-timeout sizing, and queue DLQ policy. The operational checklist is covered in Await Boundary Operations.

That matters for plugin-style side effects after an await boundary. A resumed queue-async execution can replay the remainder of the pipeline after a downstream retry, so once-only side-effect checkpointing is a separate concern from await durability itself.

Webhook Example

steps:
  - name: "Fraud Check"
    kind: "await"
    cardinality: "ONE_TO_ONE"
    input: "com.example.FraudCheckRequest"
    output: "com.example.FraudCheckDecision"
    timeout: "PT10M"
    idempotencyKeyFields: ["orderId"]
    await:
      correlation:
        strategy: "signedResumeToken"
      transport:
        type: "webhook"
        request:
          url: "https://partner.example/fraud-check"
        callback:
          baseUrl: "https://orchestrator.example"

Webhook dispatch sends an envelope containing the interaction id, correlation id, resume token, deadline, request payload, tenant id, step id, output type, and callback metadata when configured. Completion is submitted through the generated REST/gRPC completion APIs; TPF validates the token before accepting the response snapshot.

Kafka Example

steps:
  - name: "Brokered Fraud Check"
    kind: "await"
    cardinality: "ONE_TO_ONE"
    input: "com.example.FraudCheckRequest"
    output: "com.example.FraudCheckDecision"
    timeout: "PT10M"
    idempotencyKeyFields: ["orderId"]
    await:
      correlation:
        strategy: "signedResumeToken"
      transport:
        type: "kafka"
        request:
          topic: "fraud-check.requests"
          key: "correlationId" # optional: interactionId or correlationId
        response:
          topic: "fraud-check.decisions"
        consumer:
          group: "fraud-check-orchestrator" # optional; channel config remains authoritative
        headers:
          x-source: "tpf"

Kafka dispatch sends a framework-owned JSON envelope containing tenant id, execution id, interaction id, correlation id, step id, deadline, input/output types, resume token, request payload, and dispatch metadata. The response envelope is consumed by TPF and completed directly through AwaitCoordinator, not by looping back through REST. Use the generated REST/gRPC completion APIs for human/UI or webhook clients that are not broker consumers.

steps:
  - name: "Await Payment Provider"
    kind: "await"
    cardinality: "ONE_TO_ONE"
    input: "org.pipelineframework.csv.common.domain.PaymentRecord"
    output: "org.pipelineframework.csv.common.domain.PaymentStatus"
    timeout: "PT5M"
    idempotencyKeyFields: ["csvId", "recipient", "amount", "currency"]
    await:
      correlation:
        strategy: "signedResumeToken"
      transport:
        type: "kafka"
        request:
          topic: "csv-payments.payment.requests"
          key: "correlationId"
        response:
          topic: "csv-payments.payment.results"

Add the Quarkus Kafka messaging extension to the application that hosts the orchestrator, enable the default Kafka bridge with tpf.await.kafka.reactive-messaging.enabled=true, then configure the SmallRye channels:

tpf.await.kafka.reactive-messaging.enabled=true
mp.messaging.outgoing.tpf-await-kafka-requests.connector=smallrye-kafka
mp.messaging.outgoing.tpf-await-kafka-requests.value.serializer=org.apache.kafka.common.serialization.StringSerializer
mp.messaging.incoming.tpf-await-kafka-responses.connector=smallrye-kafka
mp.messaging.incoming.tpf-await-kafka-responses.topic=csv-payments.payment.results
mp.messaging.incoming.tpf-await-kafka-responses.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

SQS Example

steps:
  - name: "Brokered Fraud Check"
    kind: "await"
    cardinality: "ONE_TO_ONE"
    input: "com.example.FraudCheckRequest"
    output: "com.example.FraudCheckDecision"
    timeout: "PT10M"
    idempotencyKeyFields: ["orderId"]
    await:
      correlation:
        strategy: "signedResumeToken"
      transport:
        type: "sqs"
        request:
          queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789012/fraud-check-requests"
        response:
          queueUrl: "https://sqs.us-east-1.amazonaws.com/123456789012/fraud-check-decisions"

SQS dispatch sends a framework-owned JSON envelope with the same interaction identity and resume token fields as Kafka. The coordinator-side SQS completion poller consumes the response queue and admits completions through the await coordinator. SQS await v1 supports standard queues, not FIFO queue URLs. This is separate from SQS work dispatch, SQS DLQ publication, and the SQS transition-worker request/reply protocol.

tpf.await.sqs.poller.enabled=true
tpf.await.sqs.response-queue-url=https://sqs.us-east-1.amazonaws.com/123456789012/fraud-check-decisions
pipeline.orchestrator.sqs.region=us-east-1

pipeline.orchestrator.resume-token-secret must be stable for the lifetime of outstanding webhook, Kafka, and SQS interactions. If pipeline.orchestrator.resume-token-secret is missing, signed dispatch and token validation fail with a clear error rather than allowing insecure or unsigned resumptions.