Use messaging and events

Stable — Production-ready with stable APIs and full test coverage.

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

Consuming lifecycle events from other applications

External applications (monitoring dashboards, audit logs, data warehouses) can consume lifecycle events from the flow-lifecycle-out topic. SmallRye Reactive Messaging provides automatic CloudEvent handling via CloudEventMetadata.

Per-message consumption

For moderate throughput scenarios (hundreds of events per second), use per-message consumption:

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 LifecycleEventConsumer {

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

        String type = ce.getType();  // e.g., "io.serverlessworkflow.workflow.started.v1"
        String instanceId = ce.<String>getExtension("flowinstanceid").orElse(null);
        String taskId = ce.<String>getExtension("flowtaskid").orElse(null);
        String data = msg.getPayload();  // JSON event data

        // Process event (e.g., write to database, update dashboard)
        Log.infof("Lifecycle event: type=%s, instance=%s, task=%s", type, instanceId, taskId);

        return msg.ack();
    }
}

Channel configuration:

mp.messaging.incoming.lifecycle-consumer.connector=smallrye-kafka
mp.messaging.incoming.lifecycle-consumer.topic=flow-lifecycle-out
mp.messaging.incoming.lifecycle-consumer.group.id=my-consumer-group

Batch consumption for high-throughput scenarios

For high-throughput environments (thousands of events per second from many concurrent workflows), use batch consumption to reduce database transaction overhead:

import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletionStage;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;

import jakarta.enterprise.context.ApplicationScoped;

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

import io.cloudevents.CloudEvent;
import io.cloudevents.core.builder.CloudEventBuilder;
import io.cloudevents.core.provider.EventFormatProvider;
import io.cloudevents.jackson.JsonFormat;
@ApplicationScoped
public class LifecycleEventBatchConsumer {

    @Incoming("lifecycle-batch-consumer")
    public CompletionStage<Void> consumeBatch(Message<ConsumerRecords<String, String>> records) {
        List<CloudEvent> events = new ArrayList<>();

        for (ConsumerRecord<String, String> record : records.getPayload()) {
            CloudEvent ce = reconstructCloudEvent(record);  (1)
            events.add(ce);
        }

        // Process batch in single database transaction
        processBatch(events);

        return records.ack();
    }

    private CloudEvent reconstructCloudEvent(ConsumerRecord<String, String> record) {
        // Binary mode: CE attributes in Kafka headers (ce_type, ce_time, etc.)
        if (record.headers().lastHeader("ce_type") != null) {
            return buildFromBinaryMode(record);
        } else {
            // Structured mode: full CloudEvent JSON in message body
            return deserializeStructuredMode(record.value());
        }
    }

    private CloudEvent buildFromBinaryMode(ConsumerRecord<String, String> record) {
        String type = getHeader(record, "ce_type");
        String source = getHeader(record, "ce_source");
        String id = getHeader(record, "ce_id");

        CloudEventBuilder builder = CloudEventBuilder.v1()
                .withType(type)
                .withSource(URI.create(source))
                .withId(id)
                .withData("application/json", record.value().getBytes(StandardCharsets.UTF_8));

        var timeHeader = record.headers().lastHeader("ce_time");
        if (timeHeader != null) {
            builder.withTime(OffsetDateTime.parse(new String(timeHeader.value(), StandardCharsets.UTF_8)));
        }

        return builder.build();
    }

    private String getHeader(ConsumerRecord<String, String> record, String name) {
        return new String(record.headers().lastHeader(name).value(), StandardCharsets.UTF_8);
    }

    private CloudEvent deserializeStructuredMode(String json) {
        JsonFormat format = (JsonFormat) EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE);
        return format.deserialize(json.getBytes(StandardCharsets.UTF_8));
    }
}
1 SmallRye’s automatic CloudEventMetadata extraction only works in per-message mode. Batch mode requires manual CloudEvent reconstruction from headers or JSON.

Channel configuration for batch mode:

mp.messaging.incoming.lifecycle-batch-consumer.connector=smallrye-kafka
mp.messaging.incoming.lifecycle-batch-consumer.topic=flow-lifecycle-out
mp.messaging.incoming.lifecycle-batch-consumer.group.id=my-batch-consumer-group
mp.messaging.incoming.lifecycle-batch-consumer.batch=true  (1)
1 Enables batch consumption. SmallRye will invoke your method with ConsumerRecords containing multiple Kafka records per poll.

Per-message vs Batch mode trade-offs:

  • Per-message mode: Automatic CloudEventMetadata extraction via SmallRye, simpler code, adequate for moderate throughput.

  • Batch mode: Manual CloudEvent reconstruction required, higher throughput (~1000x fewer database transactions), essential for environments with many concurrent workflows generating thousands of events per second.

The overhead of manual CloudEvent reconstruction is negligible compared to database operations, making batch mode the correct choice for high-volume scenarios.

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