Flamme in Practice: Can Quarkus Topology Become a Runtime Decision?

A hands-on adoption check of runtime placement and replica behavior, including what happens when a component or the broker fails.

Aug 21, 2026

I found this little gem in the r/Quarkus community. Someone asked to test it and send feedback. That is the right call to action for me. So here we are!

Flamme is a Quarkus extension from Amadeus. It lets you write components without deciding where they will run. Put two components in the same process and they communicate through a local broker. Move one component to another process and Flamme routes the events through NATS. Protocol Buffers take care of the payload on the remote path. This smells like an interesting solution to some problems customers encounter when they modernize their heritage applications. Yep, I do not like the term legacy.

Moving code out of a monolith usually means adding a network API and a client. Then you own serialization and error handling. You also get another deployment contract. Flamme tries to keep the component code stable while the topology changes around it.

I wanted to see the real behavior. Happy-path diagrams or posts on Reddit are nice, but I really want something runnable to test promises. Throw an exception inside a remote component and start two worker replicas to see whether NATS shares the work or sends it to both.

This became a hands-on field test of Flamme 1.0.0-SNAPSHOT. We will use one pinned commit because there is no published release artifact yet. The result is a small Quarkus application that runs in two different topologies from the same JAR.

What We Build

We will build Release Gate. It evaluates a release candidate through four Flamme components:

The validator checks the candidate. The risk scorer calculates a score from the number of changed files and critical dependencies. The decider approves low-risk releases and sends the result back to the gateway.

The risk scorer also accepts a delay. That gives us a component that may become expensive enough to move away from the API process. Each response contains processedBy and decidedBy, so we can see where the work ran without adding a tracing backend.

First, all four components run in one JVM. Then we start the same JAR twice and move only the risk scorer to the second process:

The application source stays the same during that move. We only change runtime properties.

What You Need

The commands below use the versions I tested. Podman runs NATS and the test container.

You need JDK 21 on PATH.
You need Podman 5 or later with a running machine or socket.
You need curl.
You need three terminals for the split topology and replica test.
Plan for about ☕️☕️☕️☕️☕️ if you want to run every failure case.

On macOS or Windows, start the Podman machine first:

podman machine start

I ran the example with Java 21 and Quarkus 3.34.1. NATS is pinned to 2.14.1. The Flamme commit is 8afdaf6e8b59bc3b443750cf099971593ddb66c9.

Get the Project

Flamme currently uses the version 1.0.0-SNAPSHOT. I could not point the demo at a released Maven artifact because there is none. The repository therefore includes the Flamme runtime and deployment modules from the pinned commit.

Clone the Main Thread repository and enter the example:

git clone https://github.com/myfear/the-main-thread.git
cd the-main-thread/flamme-release-gate

The Maven reactor has three modules:

<modules>
    <module>vendor/flamme/runtime</module>
    <module>vendor/flamme/deployment</module>
    <module>app</module>
</modules>

The Java sources under vendor/flamme match the pinned Flamme commit. Only the two module POM files use this demo as their parent. That keeps the test reproducible while Flamme is still a snapshot.

The application module three four Quarkus extensions:

quarkus-rest-jackson provides the JSON endpoint.
quarkus-grpc generates Java classes from the protobuf file, as described in the Quarkus gRPC guide. NATS remains the transport.
quarkus-hibernate-validator validates the HTTP request.

The Flamme dependency comes from the vendored reactor module:

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-rest-jackson</artifactId>
</dependency>
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-grpc</artifactId>
</dependency>
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-arc</artifactId>
</dependency>
<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-hibernate-validator</artifactId>
</dependency>
<dependency>
    <groupId>com.amadeus</groupId>
    <artifactId>flamme</artifactId>
    <version>1.0.0-SNAPSHOT</version>
</dependency>

This setup is slightly more involved than the usual Quarkus project creation command you see in my tutorials. But maybe Flamme get’s a release soon.

Define the Protobuf Payload

Flamme passes a Map<String, Message> between components. Each value is a protobuf message. Local components receive the map through memory. Remote components receive a serialized copy over NATS.

Create app/src/main/proto/release_gate.proto:

syntax = "proto3";

option java_multiple_files = true;
option java_package = "com.themainthread.releasegate.proto";

message ReleaseCandidate {
  string id = 1;
  int32 changed_files = 2;
  int32 critical_dependencies = 3;
  bool force_risk_failure = 4;
  int32 analysis_delay_millis = 5;
}

message ReleaseAssessment {
  int32 score = 1;
  string summary = 2;
  string processed_by = 3;
}

message ReleaseDecision {
  bool approved = 1;
  string reason = 2;
  string decided_by = 3;
}

ReleaseCandidate enters the pipeline. ReleaseAssessment appears after scoring. ReleaseDecision is the final result. The processed_by and decided_by fields exist for our topology checks.

The failure flag also belongs to the candidate. It lets us break the scorer on purpose later. A deterministic failure switch in examples like this is a lot easier to understand ad test than waiting for a network problem. Trying hard to make my tutorials worthwile for y’all.

Add the Payload Keys and Node Configuration

Each protobuf message lives under a string key in the payload map. Create app/src/main/java/com/themainthread/releasegate/PayloadKeys.java:

package com.themainthread.releasegate;

final class PayloadKeys {

    static final String ASSESSMENT = "ASSESSMENT";
    static final String CANDIDATE = "CANDIDATE";
    static final String DECISION = "DECISION";

    private PayloadKeys() {
    }
}

We also need a name for each running process. Create app/src/main/java/com/themainthread/releasegate/ReleaseGateConfig.java:

package com.themainthread.releasegate;

import io.smallrye.config.ConfigMapping;
import io.smallrye.config.WithDefault;

@ConfigMapping(prefix = "release-gate")
interface ReleaseGateConfig {

    @WithDefault("monolith")
    String nodeId();
}

The default node is monolith. The split run changes it to api and worker-a. The replica test adds worker-b.

Declare the Flamme Components

A Flamme component starts as a Java interface with one method. The @Flamme annotation gives the component a service name. It also declares the subjects the component consumes and produces.

@MultiPayloadKey tells the remote decoder which named protobuf messages it must reconstruct. This detail matters once an edge crosses NATS.

Start with the gateway

Create app/src/main/java/com/themainthread/releasegate/ReleaseGateway.java:

package com.themainthread.releasegate;

import com.amadeus.flamme.runtime.annotations.Flamme;
import com.amadeus.flamme.runtime.annotations.Flamme.MultiPayloadKey;
import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseAssessment;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import com.themainthread.releasegate.proto.ReleaseDecision;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

@Flamme(
        serviceName = "release-gateway",
        consumes = {},
        produces = {"candidate-submitted"},
        multiPayloadKeys = {
                @MultiPayloadKey(id = PayloadKeys.CANDIDATE, type = ReleaseCandidate.class),
                @MultiPayloadKey(id = PayloadKeys.ASSESSMENT, type = ReleaseAssessment.class),
                @MultiPayloadKey(id = PayloadKeys.DECISION, type = ReleaseDecision.class)
        })
public interface ReleaseGateway {

    CompletableFuture<Map<String, Message>> evaluate(Map<String, Message> payload);
}

The gateway has no input subject because our REST resource calls it directly. It publishes candidate-submitted, then waits on a CompletableFuture.

The gateway declares all three payload keys because it decodes the final reply. This is one of the places where Flamme makes the event flow compact. The generated implementation owns the reply subject and completes the future when the terminal component answers.

Add the validator

Create app/src/main/java/com/themainthread/releasegate/CandidateValidator.java:

package com.themainthread.releasegate;

import com.amadeus.flamme.runtime.annotations.Flamme;
import com.amadeus.flamme.runtime.annotations.Flamme.MultiPayloadKey;
import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import java.util.Map;

@Flamme(
        serviceName = "candidate-validator",
        consumes = {"candidate-submitted"},
        produces = {"candidate-validated"},
        multiPayloadKeys = {
                @MultiPayloadKey(id = PayloadKeys.CANDIDATE, type = ReleaseCandidate.class)
        })
public interface CandidateValidator {

    Map<String, Message> validate(Map<String, Message> payload);
}

The validator consumes the gateway event and produces candidate-validated. At this point the payload contains only ReleaseCandidate, so that is the only key it declares.

Declare the risk scorer

Create app/src/main/java/com/themainthread/releasegate/RiskScorer.java:

package com.themainthread.releasegate;

import com.amadeus.flamme.runtime.annotations.Flamme;
import com.amadeus.flamme.runtime.annotations.Flamme.MultiPayloadKey;
import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import java.util.Map;

@Flamme(
        serviceName = "risk-scorer",
        consumes = {"candidate-validated"},
        produces = {"risk-scored"},
        multiPayloadKeys = {
                @MultiPayloadKey(id = PayloadKeys.CANDIDATE, type = ReleaseCandidate.class)
        })
public interface RiskScorer {

    Map<String, Message> score(Map<String, Message> payload);
}

This is the component we will move. Its Java contract says nothing about NATS or process boundaries. It receives a map and returns a map.

That simple method is the part that caught my attention in the first place. The location decision sits outside the business interface. Where it actually belongs.

Finish with the decider

Create app/src/main/java/com/themainthread/releasegate/ReleaseDecider.java:

package com.themainthread.releasegate;

import com.amadeus.flamme.runtime.annotations.Flamme;
import com.amadeus.flamme.runtime.annotations.Flamme.MultiPayloadKey;
import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseAssessment;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import java.util.Map;

@Flamme(
        serviceName = "release-decider",
        consumes = {"risk-scored"},
        produces = {},
        multiPayloadKeys = {
                @MultiPayloadKey(id = PayloadKeys.CANDIDATE, type = ReleaseCandidate.class),
                @MultiPayloadKey(id = PayloadKeys.ASSESSMENT, type = ReleaseAssessment.class)
        })
public interface ReleaseDecider {

    Map<String, Message> decide(Map<String, Message> payload);
}

The empty produces array marks the terminal component. Flamme sends its result to the gateway reply subject.

Before we continue, look at multiPayloadKeys. Which messages does the decider need if risk-scored crosses NATS? It needs both the original candidate and the assessment added by the scorer. Missing either declaration makes the remote payload incomplete.

There is one extra rule that I found by running the code: keep the Flamme interfaces public. Flamme invokes component methods through reflection. Package-private interfaces compiled, but the first request logged an invocation error and timed out. The current build step does not catch that visibility problem.

Implement the Components

The interfaces describe the graph and CDI beans contain the real work. Flamme finds each implementation through @FlammeImpl.

Validate the candidate

Create app/src/main/java/com/themainthread/releasegate/CandidateValidatorImpl.java:

package com.themainthread.releasegate;

import com.amadeus.flamme.runtime.annotations.FlammeImpl;
import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import io.quarkus.arc.Unremovable;
import jakarta.enterprise.context.ApplicationScoped;
import java.util.HashMap;
import java.util.Map;
import org.jboss.logging.Logger;

@ApplicationScoped
@FlammeImpl
@Unremovable
public class CandidateValidatorImpl implements CandidateValidator {

    private static final Logger LOG = Logger.getLogger(CandidateValidatorImpl.class);

    private final ReleaseGateConfig config;

    CandidateValidatorImpl(ReleaseGateConfig config) {
        this.config = config;
    }

    @Override
    public Map<String, Message> validate(Map<String, Message> payload) {
        ReleaseCandidate candidate = (ReleaseCandidate) payload.get(PayloadKeys.CANDIDATE);
        if (candidate == null || candidate.getId().isBlank()) {
            throw new IllegalArgumentException("release id must not be blank");
        }
        LOG.infov(
                "node={0} component=candidate-validator release={1}",
                config.nodeId(),
                candidate.getId());
        return new HashMap<>(payload);
    }
}

@Unremovable keeps the bean available even though application code never injects the implementation class directly. Flamme resolves it at runtime through the annotated interface.

The method returns a copy of the payload map. Protobuf messages are immutable, and each component creates a new map before it adds data. That keeps the local path closer to the remote path, where serialization already creates a new payload.

Calculate the risk

Create app/src/main/java/com/themainthread/releasegate/RiskScorerImpl.java:

package com.themainthread.releasegate;

import com.amadeus.flamme.runtime.annotations.FlammeImpl;
import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseAssessment;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import io.quarkus.arc.Unremovable;
import jakarta.enterprise.context.ApplicationScoped;
import java.util.HashMap;
import java.util.Map;
import org.jboss.logging.Logger;

@ApplicationScoped
@FlammeImpl
@Unremovable
public class RiskScorerImpl implements RiskScorer {

    private static final Logger LOG = Logger.getLogger(RiskScorerImpl.class);

    private final ReleaseGateConfig config;

    RiskScorerImpl(ReleaseGateConfig config) {
        this.config = config;
    }

    @Override
    public Map<String, Message> score(Map<String, Message> payload) {
        ReleaseCandidate candidate = (ReleaseCandidate) payload.get(PayloadKeys.CANDIDATE);
        if (candidate.getForceRiskFailure()) {
            throw new IllegalStateException("forced risk scorer failure");
        }

        delay(candidate.getAnalysisDelayMillis());
        int score = Math.min(
                100,
                candidate.getChangedFiles() * 2
                        + candidate.getCriticalDependencies() * 15);
        ReleaseAssessment assessment = ReleaseAssessment.newBuilder()
                .setScore(score)
                .setSummary(
                        score < 50
                                ? "risk stays below the release threshold"
                                : "risk exceeds the release threshold")
                .setProcessedBy(config.nodeId())
                .build();

        Map<String, Message> result = new HashMap<>(payload);
        result.put(PayloadKeys.ASSESSMENT, assessment);
        LOG.infov(
                "node={0} component=risk-scorer release={1} score={2}",
                config.nodeId(),
                candidate.getId(),
                score);
        return result;
    }

    private static void delay(int delayMillis) {
        if (delayMillis <= 0) {
            return;
        }
        try {
            Thread.sleep(delayMillis);
        } catch (InterruptedException exception) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException("risk analysis was interrupted", exception);
        }
    }
}

A simple score calculation that tracks changed files by adding two points. Each critical dependency adds 15. We cap the score at 100.

analysisDelayMillis simulates expensive work. I use a blocking sleep here because the component itself is the unit we want to move and observe. A real CPU-heavy scorer would do real something else here obviously. A real I/O-heavy scorer should use an async or reactive contract once Flamme supports that.

The failure flag will show us how an exception travels through the framework. In this snapshot, the worker log is where that journey stops.

Make the decision

Create app/src/main/java/com/themainthread/releasegate/ReleaseDeciderImpl.java:

package com.themainthread.releasegate;

import com.amadeus.flamme.runtime.annotations.FlammeImpl;
import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseAssessment;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import com.themainthread.releasegate.proto.ReleaseDecision;
import io.quarkus.arc.Unremovable;
import jakarta.enterprise.context.ApplicationScoped;
import java.util.HashMap;
import java.util.Map;
import org.jboss.logging.Logger;

@ApplicationScoped
@FlammeImpl
@Unremovable
public class ReleaseDeciderImpl implements ReleaseDecider {

    private static final int APPROVAL_THRESHOLD = 50;
    private static final Logger LOG = Logger.getLogger(ReleaseDeciderImpl.class);

    private final ReleaseGateConfig config;

    ReleaseDeciderImpl(ReleaseGateConfig config) {
        this.config = config;
    }

    @Override
    public Map<String, Message> decide(Map<String, Message> payload) {
        ReleaseCandidate candidate =
                (ReleaseCandidate) payload.get(PayloadKeys.CANDIDATE);
        ReleaseAssessment assessment =
                (ReleaseAssessment) payload.get(PayloadKeys.ASSESSMENT);
        boolean approved = assessment.getScore() < APPROVAL_THRESHOLD;
        ReleaseDecision decision = ReleaseDecision.newBuilder()
                .setApproved(approved)
                .setReason(
                        approved
                                ? "approved for release"
                                : "manual review required")
                .setDecidedBy(config.nodeId())
                .build();

        Map<String, Message> result = new HashMap<>(payload);
        result.put(PayloadKeys.DECISION, decision);
        LOG.infov(
                "node={0} component=release-decider release={1} approved={2}",
                config.nodeId(),
                candidate.getId(),
                approved);
        return result;
    }
}

A score below 50 is approved. The terminal result contains the node ID, then Flamme sends the complete map back to the gateway.

The decider does not know whether the assessment came from another method call or another machine. That is the location transparency we want to test.

Add the REST Boundary

The REST endpoint turns JSON into ReleaseCandidate. It calls the generated gateway and maps the completed payload back to JSON.

Create app/src/main/java/com/themainthread/releasegate/ReleaseResource.java:

package com.themainthread.releasegate;

import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseAssessment;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import com.themainthread.releasegate.proto.ReleaseDecision;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.validation.Valid;
import jakarta.validation.constraints.Min;
import jakarta.validation.constraints.NotBlank;
import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.MediaType;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

@ApplicationScoped
@Path("/releases")
@Consumes(MediaType.APPLICATION_JSON)
@Produces(MediaType.APPLICATION_JSON)
public class ReleaseResource {

    private final ReleaseGateway gateway;

    ReleaseResource(ReleaseGateway gateway) {
        this.gateway = gateway;
    }

    @POST
    @Path("/evaluate")
    public CompletableFuture<ReleaseResponse> evaluate(
            @Valid ReleaseRequest request) {
        ReleaseCandidate candidate = ReleaseCandidate.newBuilder()
                .setId(request.id())
                .setChangedFiles(request.changedFiles())
                .setCriticalDependencies(request.criticalDependencies())
                .setForceRiskFailure(request.forceRiskFailure())
                .setAnalysisDelayMillis(request.analysisDelayMillis())
                .build();
        Map<String, Message> payload = new HashMap<>();
        payload.put(PayloadKeys.CANDIDATE, candidate);

        return gateway.evaluate(payload).thenApply(this::toResponse);
    }

    private ReleaseResponse toResponse(Map<String, Message> payload) {
        ReleaseCandidate candidate =
                (ReleaseCandidate) payload.get(PayloadKeys.CANDIDATE);
        ReleaseAssessment assessment =
                (ReleaseAssessment) payload.get(PayloadKeys.ASSESSMENT);
        ReleaseDecision decision =
                (ReleaseDecision) payload.get(PayloadKeys.DECISION);
        return new ReleaseResponse(
                candidate.getId(),
                assessment.getScore(),
                decision.getApproved(),
                decision.getReason(),
                assessment.getProcessedBy(),
                decision.getDecidedBy());
    }

    public record ReleaseRequest(
            @NotBlank String id,
            @Min(0) int changedFiles,
            @Min(0) int criticalDependencies,
            boolean forceRiskFailure,
            @Min(0) int analysisDelayMillis) {
    }

    public record ReleaseResponse(
            String releaseId,
            int riskScore,
            boolean approved,
            String reason,
            String processedBy,
            String decidedBy) {
    }
}

The endpoint returns the CompletableFuture directly. Quarkus REST writes the response when the Flamme gateway completes it.

Validation stays at the HTTP boundary. A missing id returns HTTP 400 before the request enters the event graph. The validator still checks the protobuf payload because another caller could invoke the gateway without using REST.

Configure Flamme

Create app/src/main/resources/application.properties:

flamme.nats.url=${NATS_URL:nats://localhost:4222}
flamme.nats.connection-name=release-gate
flamme.reply-timeout=3

release-gate.node-id=${RELEASE_GATE_NODE_ID:monolith}

quarkus.grpc.server.use-separate-server=false
quarkus.otel.sdk.disabled=true

flamme.nats.url points at the broker. An environment variable can replace the local URL when the broker runs elsewhere.

flamme.reply-timeout uses seconds. Three seconds keeps our failure tests short. Use the latency budget of the real operation in production. A large value keeps HTTP requests and reply futures open while a worker or broker is unavailable.

release-gate.node-id only helps us see placement. A real service can use the pod name or another stable instance identifier.

The gRPC extension generates the protobuf classes. We let its server share the HTTP port so every process needs only one port. Flamme also brings in OpenTelemetry. This example has no collector, so the SDK is disabled. Remove that property when you configure an exporter.

The local broker uses plaintext nats://. Keep port 4222 on your development machine. This Flamme snapshot builds the NATS client from the server URL and connection name only. I would want first-class credential and TLS configuration before events leave a trusted local network. But not necessary for this first little tutorial.

Build the Application

Package the application from the flamme-release-gate directory:

./mvnw package

Maven runs the tests as part of the package lifecycle. -am tells it to build the required Flamme modules in the same reactor. The runnable JAR is:

app/target/quarkus-app/quarkus-run.jar

We will use this exact JAR for every process below.

Run the Monolith

My first attempt was the obvious one. Every component was local, so I started the application without NATS. Quarkus failed during startup:

ERROR [com.amadeus.flamme.runtime.ConnectionInitializer]
there was an error connecting to NATS

Caused by: com.amadeus.flamme.runtime.errors.NatsConnectionError:
there was an error connecting to NATS

Caused by: java.io.IOException:
Unable to connect to NATS servers: [nats://localhost:4222]

Flamme creates the NATS connection unconditionally during startup. The local topology still needs a broker running, even though its events stay inside the process.

Start NATS:

podman run --rm --name flamme-release-gate-nats \
  -p 4222:4222 \
  -d nats:2.14.1-alpine

Now start the application:

java -jar app/target/quarkus-app/quarkus-run.jar

Send a release candidate from another terminal:

curl -s \
  -H 'Content-Type: application/json' \
  -d '{
    "id": "release-42",
    "changedFiles": 6,
    "criticalDependencies": 1,
    "forceRiskFailure": false,
    "analysisDelayMillis": 0
  }' \
  http://localhost:8080/releases/evaluate

The response is deterministic:

{
  "releaseId": "release-42",
  "riskScore": 27,
  "approved": true,
  "reason": "approved for release",
  "processedBy": "monolith",
  "decidedBy": "monolith"
}

The log shows all three processing components on the same node:

node=monolith component=candidate-validator release=release-42
node=monolith component=risk-scorer release=release-42 score=27
node=monolith component=release-decider release=release-42 approved=true

At this point Flamme uses its local broker. The payload map stays in memory. No protobuf encoding is needed between these components.

Move the Risk Scorer to Another Process

Stop the monolith with Ctrl+C. Keep NATS running.

Start the API process in the first terminal:

java \
  -Drelease-gate.node-id=api \
  -Dflamme.services.risk-scorer.remote=true \
  -jar app/target/quarkus-app/quarkus-run.jar

The property says that risk-scorer is remote from this process. The gateway and validator stay local. So does the decider.

Start a worker in the second terminal:

java \
  -Drelease-gate.node-id=worker-a \
  -Dflamme.services.candidate-validator.remote=true \
  -Dflamme.services.release-decider.remote=true \
  -Dquarkus.http.port=8081 \
  -jar app/target/quarkus-app/quarkus-run.jar

The worker marks the validator and decider as remote. That leaves only the risk scorer local. Port 8081 avoids an HTTP port collision. We do not call the worker’s REST endpoint.

Send a request to the API:

curl -s \
  -H 'Content-Type: application/json' \
  -d '{
    "id": "release-split",
    "changedFiles": 6,
    "criticalDependencies": 1,
    "forceRiskFailure": false,
    "analysisDelayMillis": 100
  }' \
  http://localhost:8080/releases/evaluate

This time the response shows two nodes:

{
  "releaseId": "release-split",
  "riskScore": 27,
  "approved": true,
  "reason": "approved for release",
  "processedBy": "worker-a",
  "decidedBy": "api"
}

The matching logs are:

node=api component=candidate-validator release=release-split
node=worker-a component=risk-scorer release=release-split score=27
node=api component=release-decider release=release-split approved=true

This is the Flamme promise in a form we can see. The validator ran in API memory. Flamme encoded the candidate and published candidate-validated to NATS. The worker rebuilt the declared protobuf payload and ran the scorer. Then risk-scored crossed NATS in the other direction, and the API decider completed the gateway future.

We changed process placement with properties. The component interfaces and implementations did not change. The packaged JAR did not change either.

Add a Second Worker

Now let us test an easy production assumption. If we start another risk worker, will NATS load-balance the requests?

Start worker-b in a third terminal:

java \
  -Drelease-gate.node-id=worker-b \
  -Dflamme.services.candidate-validator.remote=true \
  -Dflamme.services.release-decider.remote=true \
  -Dquarkus.http.port=8082 \
  -jar app/target/quarkus-app/quarkus-run.jar

Send one request with a 500 ms delay:

curl -s \
  -H 'Content-Type: application/json' \
  -d '{
    "id": "release-replicas",
    "changedFiles": 8,
    "criticalDependencies": 1,
    "forceRiskFailure": false,
    "analysisDelayMillis": 500
  }' \
  http://localhost:8080/releases/evaluate

The API returns the first completed result:

{
  "releaseId": "release-replicas",
  "riskScore": 31,
  "approved": true,
  "reason": "approved for release",
  "processedBy": "worker-a",
  "decidedBy": "api"
}

Both workers processed the same release:

node=worker-a component=risk-scorer release=release-replicas score=31
node=worker-b component=risk-scorer release=release-replicas score=31

The API also ran the decider twice:

node=api component=release-decider release=release-replicas approved=true
node=api component=release-decider release=release-replicas approved=true

The current NATS transport uses a plain subject subscription. Every subscriber gets the event. There is no queue group that lets several workers share messages.

This can be correct for broadcast events. It is a problem for CPU work where one event should run once. Payments and emails make the duplicate even more visible. Component handlers need to be idempotent until Flamme supports explicit competing-consumer semantics.

Stop worker-b with Ctrl+C before you continue. Keep the API and worker-a running.

Break the Remote Component

The next request asks the scorer to throw:

curl -s \
  -H 'Content-Type: application/json' \
  -d '{
    "id": "release-failure",
    "changedFiles": 8,
    "criticalDependencies": 1,
    "forceRiskFailure": true,
    "analysisDelayMillis": 0
  }' \
  http://localhost:8080/releases/evaluate

The worker logs:

error invoking com.themainthread.releasegate.RiskScorer

After three seconds, the API returns HTTP 500:

500 - Internal Server Error
java.util.concurrent.TimeoutException

I expected the gateway future to complete with a component error. The current handler catches the invocation error and only writes a log message. It does not publish an error reply. The gateway waits until flamme.reply-timeout expires.

There is already code on the reply side that can decode an error payload. The missing part is sending that payload when a handler fails. A stable error envelope would make this much easier to operate. It should name the component and provide a stable error code. It also needs a correlation ID. A serialized Java stack trace would only move the mess across the network.

Stop NATS During a Request

Keep the API and worker-a running, then stop the broker:

podman stop flamme-release-gate-nats

Send another request to port 8080. The validator still runs because it is local:

node=api component=candidate-validator release=release-no-broker

The remote stage never receives the event. Three seconds later, the caller gets the same TimeoutException.

The NATS client starts reconnecting and logs Connection refused roughly every two seconds. Flamme’s broker catches publish failures without completing the gateway future or logging the failed subject. From the caller’s view, a lost publish and a crashed component look the same.

The transport uses Core NATS publish and subscribe. There is no durable stream or acknowledgement. There is also no replay or dead-letter path. That gives the current remote path an at-most-once delivery boundary. A message may disappear when no subscriber is active or when the broker connection drops at the wrong time.

What I Would Change

The main idea worked. I could move the scorer from memory to NATS without touching its Java code. The experiment also left me with a short list of ideas for the next Flamme version.

Let local mode start without NATS

ConnectionInitializer connects to NATS on every startup. I would initialize the transport only when the resolved graph has a remote edge. An explicit flamme.nats.enabled=false property would also help, as long as startup fails when the topology still needs NATS.

That change would make the local mode match its own operational story. A one-process application could run without broker infrastructure.

Send component failures back to the gateway

The gateway already has a reply future and the codec can recognize an error entry. The handler should publish a structured error envelope to replyTo when an implementation throws.

The HTTP layer could then map a worker failure differently from a timeout. Operators would also get the component name and correlation ID without reading logs from every replica.

Make replica semantics explicit

Flamme should let each component choose between broadcast and work sharing. NATS queue groups provide the second option. A property such as flamme.services.risk-scorer.queue-group=release-risk would make the intent visible.

I would keep broadcast as a supported mode because event listeners often need it. Worker components often do not.

Catch reflection problems during the build

Quarkus extensions have a good place to validate component metadata during augmentation. Flamme already checks method signatures. It could also reject an inaccessible interface or implementation and name the exact type in the build error.

A build failure is much cheaper than a request that logs error invoking ... and waits for a timeout.

State the delivery contract

The documentation should say that the current NATS transport is at most once. Teams need that fact when they design retries. It also decides where idempotency belongs and which work is safe inside a component.

The transport abstraction also leaves room for Kafka or JetStream. That would be a valuable contribution, but the semantics need to stay explicit. Kafka adds consumer groups and replay. JetStream adds persistence and acknowledgements. Those features also add ownership decisions that core NATS currently avoids.

Test the Application

The manual run proves process placement. Automated tests cover the local pipeline and request validation. They also check the risk calculation and forced scorer failure.

Stop the API and worker-a before you run the suite again. This avoids port conflicts and gives the tests a clean broker state.

Add the test dependencies to app/pom.xml:

<dependency>
    <groupId>io.quarkus</groupId>
    <artifactId>quarkus-junit</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>io.rest-assured</groupId>
    <artifactId>rest-assured</artifactId>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>io.github.amadeusitgroup.testcontainers</groupId>
    <artifactId>nats</artifactId>
    <version>1.0.9</version>
    <scope>test</scope>
</dependency>

Start NATS for a Quarkus test

The REST test uses a Quarkus test resource. Create app/src/test/java/com/themainthread/releasegate/NatsTestResource.java:

package com.themainthread.releasegate;

import io.github.amadeusitgroup.testcontainers.nats.NatsContainer;
import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
import java.util.Map;

public class NatsTestResource
        implements QuarkusTestResourceLifecycleManager {

    private static final int NATS_PORT = 4222;

    private NatsContainer container;

    @Override
    public Map<String, String> start() {
        container = new NatsContainer("nats:2.14.1-alpine")
                .withExposedPorts(NATS_PORT);
        container.start();
        return Map.of(
                "flamme.nats.url",
                "nats://localhost:" + container.getMappedPort(NATS_PORT),
                "release-gate.node-id",
                "test-node");
    }

    @Override
    public void stop() {
        if (container != null) {
            container.stop();
        }
    }
}

The test container uses a random host port, then passes the mapped NATS URL into Quarkus. This also documents the current startup dependency on NATS.

Test the REST path

Create app/src/test/java/com/themainthread/releasegate/ReleaseResourceTest.java:

package com.themainthread.releasegate;

import static io.restassured.RestAssured.given;
import static org.hamcrest.Matchers.equalTo;

import io.quarkus.test.common.QuarkusTestResource;
import io.quarkus.test.junit.QuarkusTest;
import io.restassured.http.ContentType;
import org.junit.jupiter.api.Test;

@QuarkusTest
@QuarkusTestResource(
        value = NatsTestResource.class,
        restrictToAnnotatedClass = true)
class ReleaseResourceTest {

    @Test
    void evaluatesReleaseThroughTheLocalPipeline() {
        given()
                .contentType(ContentType.JSON)
                .body("""
                        {
                          "id": "release-42",
                          "changedFiles": 6,
                          "criticalDependencies": 1,
                          "forceRiskFailure": false,
                          "analysisDelayMillis": 0
                        }
                        """)
                .when()
                .post("/releases/evaluate")
                .then()
                .statusCode(200)
                .body("releaseId", equalTo("release-42"))
                .body("riskScore", equalTo(27))
                .body("approved", equalTo(true))
                .body("processedBy", equalTo("test-node"))
                .body("decidedBy", equalTo("test-node"));
    }

    @Test
    void rejectsARequestWithoutAReleaseId() {
        given()
                .contentType(ContentType.JSON)
                .body("""
                        {
                          "changedFiles": 6,
                          "criticalDependencies": 1,
                          "forceRiskFailure": false,
                          "analysisDelayMillis": 0
                        }
                        """)
                .when()
                .post("/releases/evaluate")
                .then()
                .statusCode(400);
    }
}

The first test checks the complete local event graph through HTTP. The second catches a mistake I made during the manual run: I sent the wrong JSON field and reached protobuf construction with a null ID. Keeping validation at the boundary turns that into HTTP 400.

Test the scorer in isolation

Create app/src/test/java/com/themainthread/releasegate/RiskScorerTest.java:

package com.themainthread.releasegate;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;

import com.google.protobuf.Message;
import com.themainthread.releasegate.proto.ReleaseAssessment;
import com.themainthread.releasegate.proto.ReleaseCandidate;
import java.util.Map;
import org.junit.jupiter.api.Test;

class RiskScorerTest {

    private final RiskScorerImpl scorer =
            new RiskScorerImpl(() -> "unit-test");

    @Test
    void calculatesDeterministicRisk() {
        ReleaseCandidate candidate = ReleaseCandidate.newBuilder()
                .setId("release-42")
                .setChangedFiles(6)
                .setCriticalDependencies(1)
                .build();

        Map<String, Message> result =
                scorer.score(Map.of(PayloadKeys.CANDIDATE, candidate));

        ReleaseAssessment assessment =
                (ReleaseAssessment) result.get(PayloadKeys.ASSESSMENT);
        assertEquals(27, assessment.getScore());
        assertEquals("unit-test", assessment.getProcessedBy());
    }

    @Test
    void surfacesForcedRiskFailure() {
        ReleaseCandidate candidate = ReleaseCandidate.newBuilder()
                .setId("release-failure")
                .setForceRiskFailure(true)
                .build();

        assertThrows(
                IllegalStateException.class,
                () -> scorer.score(
                        Map.of(PayloadKeys.CANDIDATE, candidate)));
    }
}

Run all tests:

./mvnw test

The application module reports:

Tests run: 4, Failures: 0, Errors: 0, Skipped: 0
BUILD SUCCESS

These tests prove the deterministic code and the one-process graph. The two-process placement still needs the manual run because a single @QuarkusTest JVM cannot prove which external process handled the NATS event.

Clean Up

Stop each Java process with Ctrl+C. If NATS is still running, remove the container:

podman stop flamme-release-gate-nats

The container was started with --rm, so Podman removes it after the stop.

Conclusion

Flamme moved a Quarkus component from local memory to NATS through configuration. The business code stayed unchanged, and that part worked well. Before I would use the snapshot for critical work, I need optional broker startup and error replies. I also need explicit replica and delivery semantics.

Discussion about this post

Ready for more?