Testing the Messaging Bridge Pattern with Citrus
Enterprise systems rarely run on a single messaging platform. One team standardizes on Apache Kafka for its event streams, while a partner organization publishes orders through Apache Pulsar. An acquired company may send events via RabbitMQ. Another subsystem might still use JMS queues. In each case, messages produced on one messaging system need to flow into another — reliably, transparently, and without forcing either side to change its infrastructure.
This is the Messaging Bridge pattern, described in Enterprise Integration Patterns by Hohpe and Woolf. A Messaging Bridge connects two messaging systems so that messages available on one are also available on the other. It differs from a Channel Adapter, which connects a non-messaging application (like a REST API or a database) to a messaging system. A bridge connects two systems that are both messaging-aware, translating between their protocols, header formats, and delivery guarantees.
Apache Camel makes implementing a Messaging Bridge deceptively simple — a single route with a messaging from() and a messaging to().
But the simplicity of the implementation belies the complexity of testing it.
A bridge test must send a message into one messaging system, wait for it to traverse the bridge, and then verify it appears on the other messaging system — with both brokers running simultaneously, each with its own protocol and client libraries.
This is where Citrus comes in. Citrus can produce and consume messages across different messaging technologies in the same test, manage multi-broker infrastructure with Testcontainers, and validate that a message arriving on the destination system matches what was sent on the source. In this post, we walk through a complete example that tests a bidirectional Messaging Bridge between Pulsar and Kafka, using Citrus on Quarkus, Spring Boot, and the Camel YAML DSL.
The Camel route under test
The scenario models a partnership integration.
A partner organization runs on Apache Pulsar and publishes orders to a partner.orders.placed topic.
Your domain runs on Kafka and expects orders on the eip.orders.placed topic.
The Messaging Bridge consumes orders from the Pulsar topic and forwards them to the Kafka topic.
A second bridge route handles the reverse direction: shipping events published to a Kafka topic eip.shipping.scheduled are forwarded to a Pulsar topic of the same name, so the partner system can track shipment progress.
Here is the Camel route on Quarkus:
@ApplicationScoped
public class MessagingBridgeRoute extends RouteBuilder {
@Override
public void configure() {
// Pulsar → Kafka: partner orders arrive on Pulsar, bridged into Kafka
from("pulsar:persistent://public/default/partner.orders.placed"
+ "?subscriptionName=kafka-bridge"
+ "&subscriptionType=Exclusive")
.routeId("messaging-bridge-pulsar-to-kafka")
.log("Messaging Bridge — Pulsar → Kafka: ${body}")
.to("kafka:eip.orders.placed?brokers={{kafka.brokers}}")
.log("Messaging Bridge — order forwarded to Kafka");
// Kafka → Pulsar: shipping events bridged to Pulsar for partner systems
from("kafka:eip.shipping.scheduled?brokers={{kafka.brokers}}&groupId=pulsar-bridge")
.routeId("messaging-bridge-kafka-to-pulsar")
.log("Messaging Bridge — Kafka → Pulsar: ${body}")
.to("pulsar:persistent://public/default/eip.shipping.scheduled"
+ "?producerName=kafka-bridge")
.log("Messaging Bridge — shipping event forwarded to Pulsar");
}
}
The route looks remarkably simple — each bridge direction is just a from() and a to() — but there is important behavior beneath the surface.
First, the Pulsar consumer uses subscriptionType=Exclusive, which means only one consumer instance can subscribe at a time.
This prevents duplicate bridging when multiple application instances are running — only one instance receives each message.
Second, the subscriptionName=kafka-bridge creates a named, durable subscription on Pulsar.
If the bridge restarts, it resumes from where it left off rather than replaying from the beginning of the topic or losing unprocessed messages.
Third, and most critically for reliability: Camel’s synchronous processing model is what makes the bridge safe.
The to("kafka:...") call blocks until the Kafka producer receives an acknowledgement from the broker.
Only after Kafka confirms receipt does the Pulsar consumer acknowledge its message.
If the Kafka publish fails, the Pulsar message is not acknowledged, so it will be redelivered on the next attempt.
This is the essential guarantee of a Messaging Bridge — no message is lost in transit between the two systems.
On Spring Boot, the route is identical except for the annotation:
@Component
public class MessagingBridgeRoute extends RouteBuilder {
@Override
public void configure() {
// Pulsar → Kafka: partner orders arrive on Pulsar, bridged into Kafka
from("pulsar:persistent://public/default/partner.orders.placed"
+ "?subscriptionName=kafka-bridge"
+ "&subscriptionType=Exclusive")
.routeId("messaging-bridge-pulsar-to-kafka")
.log("Messaging Bridge — Pulsar → Kafka: ${body}")
.to("kafka:eip.orders.placed?brokers={{kafka.brokers}}")
.log("Messaging Bridge — order forwarded to Kafka");
// Kafka → Pulsar: shipping events bridged to Pulsar for partner systems
from("kafka:eip.shipping.scheduled?brokers={{kafka.brokers}}&groupId=pulsar-bridge")
.routeId("messaging-bridge-kafka-to-pulsar")
.log("Messaging Bridge — Kafka → Pulsar: ${body}")
.to("pulsar:persistent://public/default/eip.shipping.scheduled"
+ "?producerName=kafka-bridge")
.log("Messaging Bridge — shipping event forwarded to Pulsar");
}
}
The same routing logic expressed in the Camel YAML DSL:
# Pulsar → Kafka: partner orders arrive on Pulsar, bridged into Kafka
- route:
id: messaging-bridge-pulsar-to-kafka
from:
uri: "pulsar:persistent://public/default/partner.orders.placed"
parameters:
subscriptionName: kafka-bridge
subscriptionType: Exclusive
steps:
- log: "Messaging Bridge — Pulsar → Kafka: ${body}"
- to:
uri: "kafka:eip.orders.placed"
parameters:
brokers: "{{kafka.brokers}}"
- log: "Messaging Bridge — order forwarded to Kafka"
# Kafka → Pulsar: shipping events bridged to Pulsar for partner systems
- route:
id: messaging-bridge-kafka-to-pulsar
from:
uri: "kafka:eip.shipping.scheduled"
parameters:
brokers: "{{kafka.brokers}}"
groupId: pulsar-bridge
steps:
- log: "Messaging Bridge — Kafka → Pulsar: ${body}"
- to:
uri: "pulsar:persistent://public/default/eip.shipping.scheduled"
parameters:
producerName: kafka-bridge
- log: "Messaging Bridge — shipping event forwarded to Pulsar"
All three variants express the same bidirectional bridge: orders flow from Pulsar to Kafka, shipping events flow from Kafka to Pulsar.
What makes testing a messaging bridge different
Testing a Messaging Bridge is fundamentally different from testing a route that stays within a single broker. With a Content-Based Router or a Splitter, you send to a Kafka topic and receive from a Kafka topic — the test operates within one messaging system. With a bridge, the test must cross the protocol boundary: send on one technology, receive on another.
This creates three distinct challenges.
Multi-broker infrastructure: the test environment must run both a Kafka broker and a Pulsar broker simultaneously, each with its own startup sequence and health checks. The test cannot begin until both brokers are healthy and the bridge route has connected to both.
Cross-protocol send and receive: the test must produce a message using Pulsar’s client protocol and consume the bridged result using Kafka’s consumer protocol (or vice versa). The Citrus framework must support both protocols in the same test case, with endpoint URIs that speak each broker’s native language.
Content fidelity across systems: the test must verify that the message content survives the bridge intact. Pulsar and Kafka have different header/property models, different serialization defaults, and different metadata. The bridge must preserve the application payload even though the envelope changes.
The message template
All bridge tests use a shared JSON template stored in src/test/resources/templates/order.json:
{
"order_id": org.citrusframework:citrus-website:pom:1.1.0,
"customer_id": "CUST-00org.citrusframework:citrus-website:pom:1.1.0",
"item_sku": "SKU-SHIP-org.citrusframework:citrus-website:pom:1.1.0",
"quantity": 1,
"amount": ${amount},
"status": "${status}",
"shipping_priority": "${priority}"
}
Citrus resolves the ${variable} placeholders at runtime from test variables.
This template is used for both sending and receiving — the same template with the same variables validates that the message content survives the bridge unchanged.
Testing the Pulsar-to-Kafka bridge
The core test sends an order to the Pulsar topic and verifies it arrives on the Kafka topic.
This exercises the messaging-bridge-pulsar-to-kafka route.
Here is the test on Quarkus:
@Nested
class MessagingBridgePulsarToKafkaTest {
@Test
public void shouldBridgeOrderFromPulsarToKafka() {
t.given(
createVariables()
.variable("id", "citrus:randomNumber(4)")
.variable("amount", 200)
.variable("status", "partner-placed")
.variable("priority", "STANDARD")
);
t.given(waitForCamelRouteStarted(
"messaging-bridge-pulsar-to-kafka", camelContext));
t.when(
camel()
.send()
.endpoint(CamelSupport.camel().endpoints()
.pulsar("persistent://public/default/partner.orders.placed")
.serviceUrl("pulsar://localhost:6650")
.producerName("citrus-test")::getRawUri)
.fork(true)
.message()
.body(Resources.create("templates/order.json"))
);
t.then(
repeatOnError()
.times(10)
.actions(
receive()
.endpoint("kafka:eip.orders.placed?consumerGroup=citrus-placed-group")
.message()
.body(Resources.create("templates/order.json"))
)
);
}
}
There is a lot happening in this test, so let’s break it down.
The given phase creates test variables — a random order ID, an amount, a status of partner-placed to indicate this order originates from the partner system, and a standard shipping priority.
Then it waits for the bridge route to be fully started and connected to both brokers.
The when phase sends the order to Pulsar.
Citrus itself does not provide an endpoint implementation for Pulsar.
This is where the combination of Citrus and Apache Camel is essential.
Citrus is able to use the Camel implementation for Pulsar to send and receive messages as part of the test.
This works for all endpoint components that Apache Camel provides, that’s fantastic.
This is the cross-protocol step — notice how the endpoint is constructed using Citrus’s Camel endpoint builder: CamelSupport.camel().endpoints().pulsar(...).
This builder constructs the full Pulsar endpoint URI with the service URL (pulsar://localhost:6650) and a producer name for the test.
The ::getRawUri method reference extracts the raw URI string that the Citrus send action needs.
The fork(true) setting is important here.
It tells Citrus to send the message asynchronously — fire the Pulsar publish and immediately proceed to the receive phase without waiting for the send to complete.
Without forking, the test would send, wait for the Pulsar acknowledgement, and then start listening on Kafka — but by that time, the bridge may have already forwarded the message.
Forking the send operation overlaps the publish and consume phases, eliminating this race condition.
The then phase consumes from the Kafka output topic using repeatOnError for resilient polling.
The test retries up to 10 times with one-second pauses, waiting for the bridged message to appear on eip.orders.placed.
The receive action uses a dedicated consumer group (citrus-placed-group) to avoid interfering with the bridge route’s own consumer groups.
The validation uses the same order.json template with the same variables that were used for sending.
This proves that the message content survived the Pulsar-to-Kafka bridge intact — the order ID, amount, status, and all other fields match exactly what was sent to Pulsar.
Testing with the YAML DSL
The Camel CLI supports running routes from YAML files and testing them with Citrus YAML test definitions. The Messaging Bridge test in YAML DSL covers both bridge directions in sequence:
name: messaging-bridge-test
description: >-
Test verifying the Messaging Bridge pattern — bidirectional
Pulsar ↔ Kafka bridging
variables:
- name: kafka.broker
value: localhost:9092
- name: order.id
value: "citrus:randomNumber(4)"
- name: order.status
value: "partner-placed"
actions:
- testcontainers:
compose:
up:
file: "_infra/compose.yaml"
- waitFor:
timeout: "25000"
http:
url: http://localhost:8090
- camel:
jbang:
run:
integration:
name: "messaging-bridge"
file: "../messaging-bridge.yaml"
systemProperties:
file: "../application.properties"
# --- Pulsar → Kafka ---
- send:
endpoint: >-
camel:pulsar:persistent://public/default/partner.orders.placed?serviceUrl=pulsar://localhost:6650&producerName=citrus-bridge-test
fork: true
message:
body:
resource:
file: "templates/order.json"
- camel:
jbang:
verify:
integration: "messaging-bridge"
logMessage: "Messaging Bridge — Pulsar → Kafka"
- camel:
jbang:
verify:
integration: "messaging-bridge"
logMessage: "order forwarded to Kafka"
# --- Kafka → Pulsar ---
- createVariables:
variables:
- name: order.id
value: "citrus:randomNumber(4)"
- name: order.status
value: "shipped"
- send:
endpoint: >-
kafka:eip.shipping.scheduled?server=${kafka.broker}
message:
body:
resource:
file: "templates/order.json"
- camel:
jbang:
verify:
integration: "messaging-bridge"
logMessage: "Messaging Bridge — Kafka → Pulsar"
- camel:
jbang:
verify:
integration: "messaging-bridge"
logMessage: "shipping event forwarded to Pulsar"
The YAML test is self-contained: it starts both brokers via Testcontainers, waits for Kafka UI readiness, launches the Camel integration with camel:jbang:run, and tests both bridge directions.
The Pulsar-to-Kafka test sends an order to the Pulsar topic with fork: true and then uses camel:jbang:verify to check the running integration’s log for the expected bridge messages.
The verify action waits until the log message appears, serving as both a synchronization point and a secondary confirmation that the bridge processed the message.
The Kafka-to-Pulsar test creates fresh variables (a new order ID with status shipped), sends to the Kafka topic, and verifies the bridge log output for the reverse direction.
Notice that the YAML test uses camel:jbang:verify on the log output rather than consuming from the Pulsar output topic — this avoids the complexity of setting up a Pulsar consumer subscription in the test itself while still confirming the bridge processed the message end-to-end.
For the test infrastructure setup, shared test utilities, runtime wiring, dependencies, and how to run the tests, see the Camel EIP examples overview page.
Key takeaways
Testing a Messaging Bridge means proving that messages cross the boundary between two messaging systems with their content intact. Unlike single-broker tests, a bridge test must manage multi-broker infrastructure, send and receive across different protocols, and handle the longer startup times that come with running multiple messaging platforms.
Here is what Citrus brings to this problem:
- Citrus & Apache Camel components — Whenever there is a transport or messaging protocol that is not supported by Citrus you can use the Apache Camel component implementation.
-
Cross-protocol send and receive — The test sends to Pulsar using the
CamelSupport.camel().endpoints().pulsar(...)endpoint builder and receives from Kafka using the standard Kafka endpoint. Citrus handles both protocols in the same test case without requiring separate client libraries. -
fork(true)for overlapping operations — Forking the send ensures the test starts listening on the destination system before the bridge finishes processing, eliminating race conditions where the bridged message arrives before the test consumer is ready. -
repeatOnErrorfor async polling — The bridge introduces latency from two broker hops (source acknowledge, destination publish).repeatOnErrorpolls the destination topic and succeeds as soon as the bridged message arrives, keeping tests fast without brittle fixed sleeps. -
Template-based content fidelity — The same
order.jsontemplate is used for both sending and receiving, proving the message payload survives the protocol translation unchanged. -
camel:jbang:verifyfor log-based confirmation — In the YAML DSL test, log verification provides a secondary confirmation that the bridge processed the message, independent of consuming from the destination topic. - Multi-runtime portability — The same test logic runs on Quarkus, Spring Boot, and the Camel CLI with YAML DSL. Only the wiring annotations and infrastructure registration change; the bridge verification logic stays identical.
The complete source code for this example is available on GitHub:
Give it a try, and let us know what you think!