Use messaging and events

Quarkus Flow Messaging bridges the workflow engine with MicroProfile Reactive Messaging (SmallRye). This allows your workflows to act as fully event-driven orchestrations: they can be triggered by events, emit events to external systems, and pause execution while listening for asynchronous callbacks.

The bridge activates automatically when a supported Quarkus Messaging connector (like Kafka or AMQP) is on the classpath.

This guide shows how to:

  • Activate the Flow ↔ Reactive Messaging bridge.

  • Map the default flow-in / flow-out channels to Kafka.

  • Use the Java DSL to emit and listen for events.

  • Enable automated lifecycle events.

  • Understand the CloudEvent correlation metadata.

Prerequisites

  • A Quarkus application with Quarkus Flow set up.

  • Quarkus Messaging on the classpath (e.g., quarkus-messaging-kafka).

  • Basic familiarity with MicroProfile Reactive Messaging configuration (mp.messaging.*).

1. Add the dependencies and activate the bridge

You do not need a separate quarkus-flow-messaging artifact. The bridge is auto-registered when any quarkus-messaging-* connector is present.

pom.xml
<dependency>
  <groupId>io.quarkus</groupId>
  <artifactId>quarkus-messaging-kafka</artifactId>
</dependency>

To activate the bridge, enable the default domain event channels in your application.properties:

quarkus.flow.messaging.defaults-enabled=true

Once enabled, the engine creates two default channels:

  • flow-in (Inbound): Consumes CloudEvents (structured or binary) to start or wake up workflows.

  • flow-out (Outbound): Publishes CloudEvents emitted by workflows via SmallRye’s OutgoingCloudEventMetadata.

Dev/test shortcut: automatic messaging configuration with DevServices

In dev and test mode, instead of manually enabling the bridge and configuring every mp.messaging.* property, you can let Quarkus Flow set up the full messaging stack automatically with a single flag:

quarkus.flow.messaging.devservices-messaging-enabled=true

This single property:

  • Enables the default messaging bridge (quarkus.flow.messaging.defaults-enabled=true)

  • Enables lifecycle events (quarkus.flow.messaging.lifecycle-enabled=true)

  • Detects the Quarkus Messaging connector on the classpath and configures the three default channels (flow-in, flow-out, flow-lifecycle-out) accordingly:

    • Kafka (quarkus-messaging-kafka) — connector=smallrye-kafka, topic=<channel name>, String serializers/deserializers on each channel (the bridge payload is the CloudEvent JSON as String)

    • AMQP (quarkus-messaging-amqp) — connector=smallrye-amqp, address=<channel name> on each channel

  • Triggers Quarkus DevServices to start an ephemeral broker for the detected connector automatically

You can override any of the injected defaults in your application.properties — the injected values are low-priority defaults.

If no supported connector — or more than one — is present, Quarkus Flow injects nothing: with an ambiguous or missing connector you must configure the mp.messaging.* channels manually. A warning is logged in that case, unless the default channels are already configured in your application.properties (then the flag quietly steps aside).

This shortcut only applies in dev and test mode. In production you must activate the bridge with quarkus.flow.messaging.defaults-enabled=true and map the channels to your broker as shown in the next section.

2. Map channels to Kafka topics

Configure MicroProfile Reactive Messaging to connect the default Flow channels to your Kafka topics.

CloudEvent encoding is handled automatically by SmallRye Reactive Messaging. The message payload carries the event data, while CE attributes (specversion, type, source, …) travel via transport headers — Kafka ce_* headers or AMQP cloudEvents:* application properties. You do not need to configure custom serializers or deserializers.

# Inbound (Listens for events to start or resume workflows)
mp.messaging.incoming.flow-in.connector=smallrye-kafka
mp.messaging.incoming.flow-in.topic=flow-in

# Outbound (Publishes events emitted by the workflow)
mp.messaging.outgoing.flow-out.connector=smallrye-kafka
mp.messaging.outgoing.flow-out.topic=flow-out

# Optional if Dev Services is off:
# kafka.bootstrap.servers=localhost:9092
The same channels work with the AMQP connector — just replace smallrye-kafka with smallrye-amqp and topic with address.

3. Emit and Listen in the Java DSL

In your Quarkus Flow definitions, you interact with these Kafka topics using two primary tasks:

  • emitJson("event.type", Pojo.class): Takes a standard Java POJO from your workflow data, wraps it in a CloudEvent envelope with the specified type, and publishes it to flow-out.

  • listen("taskName", toOne("event.type")): Pauses the workflow instance and releases the execution thread. The engine defines an Event Filter waiting for a CloudEvent of event.type to arrive on flow-in. When it does, the engine automatically correlates it to the paused instance, extracts the payload, and resumes execution.

For callback or resume flows, the listen(…​) task is only half of the story. The workflow instance ID is usually carried on the outbound event as flowinstanceid, persisted by the external system, and then sent back on the callback CloudEvent so the engine can wake the correct waiting instance. See Correlating Callbacks to a Waiting Instance.

Here is a complete example of an event-driven workflow:

package org.acme;

import static io.quarkiverse.flow.dsl.FlowDSL.*;

import java.util.Map;

import jakarta.enterprise.context.ApplicationScoped;

import io.quarkiverse.flow.Flow;
import io.quarkiverse.flow.dsl.FlowWorkflowBuilder;
import io.serverlessworkflow.api.types.Workflow;

@ApplicationScoped
public class HelloMessagingFlow extends Flow {

    @Override
    public Workflow descriptor() {
        return FlowWorkflowBuilder.workflow("hello-messaging")
                .tasks(
                        listen("waitHello", toOne("org.acme.hello.request").first()),

                        // Build a response with jq
                        set("{ message: \"Hello \" + .name }"),

                        // Emit the response event
                        emitJson("org.acme.hello.response", Map.class))
                .build();
    }
}

With this in place:

  • External producers can send CloudEvents to the flow-in topic to start or continue workflows.

  • Workflows can emit CloudEvents, which are then written to the flow-out topic.

4. Enable lifecycle events (Optional)

You can configure the engine to publish its internal state changes (e.g., when a workflow starts, suspends, or faults) to a dedicated topic. This is highly useful for building observability dashboards or audit logs.

quarkus.flow.messaging.lifecycle-enabled=true

mp.messaging.outgoing.flow-lifecycle-out.connector=smallrye-kafka
mp.messaging.outgoing.flow-lifecycle-out.topic=flow-lifecycle-out

Common lifecycle event types published to this topic include:

  • io.serverlessworkflow.workflow.started.v1

  • io.serverlessworkflow.workflow.completed.v1

  • io.serverlessworkflow.task.started.v1

  • io.serverlessworkflow.task.suspended.v1

  • io.serverlessworkflow.task.faulted.v1

5. CloudEvent Correlation Headers

For idempotency and end-to-end distributed traceability, Quarkus Flow automatically attaches correlation metadata to all emitted events via CloudEvent Extension Context Attributes.

By default, every event published to flow-out will include:

  • flowinstanceid: The ULID of the executing WorkflowInstance.

  • flowtaskid: The JSON Pointer of the specific task that emitted the event (e.g., do/0/task).

If you are writing a custom consumer downstream, use SmallRye’s CloudEventMetadata to access attributes and read the payload as a String:

import java.util.concurrent.CompletionStage;

import jakarta.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;

import io.quarkus.logging.Log;
import io.smallrye.reactive.messaging.ce.CloudEventMetadata;

@ApplicationScoped
public class FlowOutConsumer {

    @Incoming("flow-out-consumer")      (1)
    public CompletionStage<Void> consume(Message<String> msg) {
        CloudEventMetadata<?> ce = msg.getMetadata(CloudEventMetadata.class)
                .orElseThrow(() -> new IllegalArgumentException("Missing CE metadata"));

        String type = ce.getType();
        String instanceId = ce.<String>getExtension("flowinstanceid").orElse(null);
        String taskId = ce.<String>getExtension("flowtaskid").orElse(null);
        String data = msg.getPayload();     (2)

        Log.infof("CE type: %s, instance: %s, task: %s", type, instanceId, taskId);
        Log.infof("Payload: %s", data);

        return msg.ack();
    }
}
1 Use a different channel name than flow-out (which is the engine’s outbound channel). Point it at the same topic/address.
2 The payload is the event data only. CE attributes (type, source, extensions, …) are in the message metadata, not in the payload.

The matching channel configuration:

mp.messaging.incoming.flow-out-consumer.connector=smallrye-kafka
mp.messaging.incoming.flow-out-consumer.topic=flow-out
Do not use io.cloudevents.CloudEvent as the parameter type in an @Incoming method. SmallRye Reactive Messaging does not auto-deserialize the payload into a CloudEvent object — the CE attributes are carried in transport headers (Kafka ce_*, AMQP cloudEvents:*) and exposed via CloudEventMetadata. Using CloudEvent directly will cause a ClassCastException.

If you need to rename these keys to match your enterprise naming conventions, or disable them entirely:

quarkus.flow.messaging.metadata.instance-id.key=mycompany-instance-id
quarkus.flow.messaging.metadata.task-id.key=mycompany-task-id

# To disable injection entirely:
# quarkus.flow.messaging.enable-metadata-correlation=false

The same flowinstanceid metadata is what supports the callback pattern described in Correlating Callbacks to a Waiting Instance.

6. Bring your own messaging (Advanced)

If you need full control over how events are consumed and produced—for example, to route events through custom enrichment logic before they reach the engine, or to filter which events get published—you can bypass the flow-in and flow-out channels.

Provide your own CDI beans implementing the Flow interfaces:

  • Exactly one io.serverlessworkflow.impl.events.EventConsumer (for inbound events).

  • Zero or more io.serverlessworkflow.impl.events.EventPublisher (for outbound events).

When you provide these beans, the default bridge is ignored.

See also