MockServer's AsyncAPI broker mocking lets you drive Kafka, MQTT, and AMQP/RabbitMQ brokers with realistic example messages, all derived directly from an AsyncAPI 2.x or 3.x specification. For Kafka, MQTT, and AMQP/RabbitMQ, MockServer can also subscribe to broker channels to record incoming messages for verification — mirroring how HTTP requests are recorded. Kafka messages can be published and consumed as plain JSON or as Avro in the Confluent Schema Registry wire format, and MQTT supports both 3.1.1 and 5.
The fastest way to try AsyncAPI mocking: supply a spec and a Kafka bootstrap address and MockServer publishes example messages immediately.
curl -s -X PUT http://localhost:1080/mockserver/asyncapi \
-H "Content-Type: application/json" \
-d '{
"spec": {
"asyncapi": "2.6.0",
"info": { "title": "Orders API", "version": "1.0.0" },
"channels": {
"orders": {
"publish": {
"message": {
"payload": {
"type": "object",
"properties": {
"orderId": { "type": "integer" },
"status": { "type": "string", "enum": ["pending", "shipped"] }
},
"required": ["orderId"]
}
}
}
}
}
},
"brokerConfig": {
"kafkaBootstrapServers": "localhost:9092",
"publishOnLoad": true,
"consume": true
}
}'curl -s -X PUT http://localhost:1080/mockserver/asyncapi/verify \
-H "Content-Type: application/json" \
-d '{
"channel": "orders",
"count": { "atLeast": 1 }
}'
# 202 Accepted = verification passedNote: consume: true tells MockServer to subscribe and record messages that arrive on the channel — it does not record MockServer's own published examples. To make the verify step above pass against recorded/consumed messages, have another producer send to the orders channel. If you only want to confirm MockServer published its example, you can drop consume from this quickstart.
See the AsyncAPI examples in the examples/bruno/asyncapi folder. The full REST API reference, schema validation details, and MQTT configuration follow below.
Load an AsyncAPI spec and start mocking via the REST API:
The request body can be either a plain AsyncAPI spec (JSON or YAML) or a JSON wrapper with broker configuration. The Java client exposes a typed loadAsyncApi(...) helper; every other language sends the same JSON over raw HTTP to PUT /mockserver/asyncapi.
MockServerClient client = new MockServerClient("localhost", 1080);
String status = client.loadAsyncApi("{\n" +
" \"spec\": {\n" +
" \"asyncapi\": \"2.6.0\",\n" +
" \"info\": { \"title\": \"Orders API\", \"version\": \"1.0.0\" },\n" +
" \"channels\": {\n" +
" \"orders\": {\n" +
" \"publish\": {\n" +
" \"message\": {\n" +
" \"payload\": {\n" +
" \"type\": \"object\",\n" +
" \"properties\": {\n" +
" \"orderId\": { \"type\": \"integer\" },\n" +
" \"status\": { \"type\": \"string\", \"enum\": [\"pending\", \"shipped\"] }\n" +
" },\n" +
" \"required\": [\"orderId\"]\n" +
" }\n" +
" }\n" +
" }\n" +
" }\n" +
" }\n" +
" },\n" +
" \"brokerConfig\": {\n" +
" \"kafkaBootstrapServers\": \"localhost:9092\",\n" +
" \"publishOnLoad\": true,\n" +
" \"consume\": true\n" +
" }\n" +
"}");const body = {
"spec": {
"asyncapi": "2.6.0",
"info": { "title": "Orders API", "version": "1.0.0" },
"channels": {
"orders": {
"publish": {
"message": {
"payload": {
"type": "object",
"properties": {
"orderId": { "type": "integer" },
"status": { "type": "string", "enum": ["pending", "shipped"] }
},
"required": ["orderId"]
}
}
}
}
}
},
"brokerConfig": {
"kafkaBootstrapServers": "localhost:9092",
"publishOnLoad": true,
"consume": true
}
};
await fetch("http://localhost:1080/mockserver/asyncapi", {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body)
});import json
import urllib.request
body = {
"spec": {
"asyncapi": "2.6.0",
"info": {"title": "Orders API", "version": "1.0.0"},
"channels": {
"orders": {
"publish": {
"message": {
"payload": {
"type": "object",
"properties": {
"orderId": {"type": "integer"},
"status": {"type": "string", "enum": ["pending", "shipped"]}
},
"required": ["orderId"]
}
}
}
}
}
},
"brokerConfig": {
"kafkaBootstrapServers": "localhost:9092",
"publishOnLoad": True,
"consume": True
}
}
req = urllib.request.Request(
"http://localhost:1080/mockserver/asyncapi",
data=json.dumps(body).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="PUT"
)
urllib.request.urlopen(req)require 'net/http'
require 'json'
require 'uri'
body = {
"spec" => {
"asyncapi" => "2.6.0",
"info" => { "title" => "Orders API", "version" => "1.0.0" },
"channels" => {
"orders" => {
"publish" => {
"message" => {
"payload" => {
"type" => "object",
"properties" => {
"orderId" => { "type" => "integer" },
"status" => { "type" => "string", "enum" => ["pending", "shipped"] }
},
"required" => ["orderId"]
}
}
}
}
}
},
"brokerConfig" => {
"kafkaBootstrapServers" => "localhost:9092",
"publishOnLoad" => true,
"consume" => true
}
}
uri = URI('http://localhost:1080/mockserver/asyncapi')
request = Net::HTTP::Put.new(uri, 'Content-Type' => 'application/json')
request.body = body.to_json
Net::HTTP.start(uri.hostname, uri.port) { |http| http.request(request) }package main
import (
"bytes"
"net/http"
)
func main() {
body := []byte(`{
"spec": {
"asyncapi": "2.6.0",
"info": { "title": "Orders API", "version": "1.0.0" },
"channels": {
"orders": {
"publish": {
"message": {
"payload": {
"type": "object",
"properties": {
"orderId": { "type": "integer" },
"status": { "type": "string", "enum": ["pending", "shipped"] }
},
"required": ["orderId"]
}
}
}
}
}
},
"brokerConfig": {
"kafkaBootstrapServers": "localhost:9092",
"publishOnLoad": true,
"consume": true
}
}`)
req, _ := http.NewRequest("PUT",
"http://localhost:1080/mockserver/asyncapi",
bytes.NewReader(body))
req.Header.Set("Content-Type", "application/json")
http.DefaultClient.Do(req)
}using System.Net.Http;
using System.Text;
using var httpClient = new HttpClient();
var json = @"{
""spec"": {
""asyncapi"": ""2.6.0"",
""info"": { ""title"": ""Orders API"", ""version"": ""1.0.0"" },
""channels"": {
""orders"": {
""publish"": {
""message"": {
""payload"": {
""type"": ""object"",
""properties"": {
""orderId"": { ""type"": ""integer"" },
""status"": { ""type"": ""string"", ""enum"": [""pending"", ""shipped""] }
},
""required"": [""orderId""]
}
}
}
}
}
},
""brokerConfig"": {
""kafkaBootstrapServers"": ""localhost:9092"",
""publishOnLoad"": true,
""consume"": true
}
}";
await httpClient.PutAsync(
"http://localhost:1080/mockserver/asyncapi",
new StringContent(json, Encoding.UTF8, "application/json")
);use reqwest::blocking::Client;
let client = Client::new();
let body = r#"{
"spec": {
"asyncapi": "2.6.0",
"info": { "title": "Orders API", "version": "1.0.0" },
"channels": {
"orders": {
"publish": {
"message": {
"payload": {
"type": "object",
"properties": {
"orderId": { "type": "integer" },
"status": { "type": "string", "enum": ["pending", "shipped"] }
},
"required": ["orderId"]
}
}
}
}
}
},
"brokerConfig": {
"kafkaBootstrapServers": "localhost:9092",
"publishOnLoad": true,
"consume": true
}
}"#;
client.put("http://localhost:1080/mockserver/asyncapi")
.header("Content-Type", "application/json")
.body(body.to_string())
.send()
.unwrap();$json = <<<'JSON'
{
"spec": {
"asyncapi": "2.6.0",
"info": { "title": "Orders API", "version": "1.0.0" },
"channels": {
"orders": {
"publish": {
"message": {
"payload": {
"type": "object",
"properties": {
"orderId": { "type": "integer" },
"status": { "type": "string", "enum": ["pending", "shipped"] }
},
"required": ["orderId"]
}
}
}
}
}
},
"brokerConfig": {
"kafkaBootstrapServers": "localhost:9092",
"publishOnLoad": true,
"consume": true
}
}
JSON;
$ch = curl_init('http://localhost:1080/mockserver/asyncapi');
curl_setopt_array($ch, [
CURLOPT_CUSTOMREQUEST => 'PUT',
CURLOPT_POSTFIELDS => $json,
CURLOPT_HTTPHEADER => ['Content-Type: application/json; charset=utf-8'],
CURLOPT_RETURNTRANSFER => true,
]);
curl_exec($ch);
curl_close($ch);curl -v -X PUT "http://localhost:1080/mockserver/asyncapi" \
-H "Content-Type: application/json" \
-d '{
"spec": {
"asyncapi": "2.6.0",
"info": { "title": "Orders API", "version": "1.0.0" },
"channels": {
"orders": {
"publish": {
"message": {
"payload": {
"type": "object",
"properties": {
"orderId": { "type": "integer" },
"status": { "type": "string", "enum": ["pending", "shipped"] }
},
"required": ["orderId"]
}
}
}
}
}
},
"brokerConfig": {
"kafkaBootstrapServers": "localhost:9092",
"publishOnLoad": true,
"consume": true
}
}'| Field | Type | Default | Description |
|---|---|---|---|
| kafkaBootstrapServers | string | null | Kafka bootstrap servers (e.g. localhost:9092) |
| kafkaGroupId | string | mockserver-async-consumer | Consumer group ID for Kafka subscribers |
| mqttBrokerUrl | string | null | MQTT broker URL (e.g. tcp://localhost:1883) |
| amqpUri | string | null | AMQP/RabbitMQ connection URI (e.g. amqp://guest:guest@localhost:5672). Supports publish and subscribe. Channels publish/consume via the binding's exchange using the channel name (or an explicit routingKey) as the routing key; with no exchange named, the default exchange is used. A message that reaches no queue is reported as an error rather than being silently discarded by the broker. This uses publisher confirms, which are a RabbitMQ extension: RabbitMQ is the supported broker. Against an AMQP 0-9-1 broker without publisher confirms MockServer falls back to publishing without them — unroutable messages then cannot be detected — but that fallback is not covered by a live-broker test. |
| mqttClientId | string | mockserver-mqtt-pub / mockserver-mqtt-sub | MQTT client ID prefix. When set, -pub and -sub suffixes are appended for the publisher and subscriber connections respectively. When unset, the defaults mockserver-mqtt-pub (publisher) and mockserver-mqtt-sub (subscriber) are used. |
| mqttQos | int | 1 | MQTT QoS level (0, 1, or 2) |
| mqttProtocolVersion | int | 3 | MQTT protocol version: 3 (3.1.1) or 5. MQTT 5 additionally delivers message headers (e.g. correlation IDs) as user properties, which MQTT 3 cannot carry. |
| kafkaValueFormat | string | json | Kafka message value format: json or avro. With avro, MockServer publishes/consumes in the Confluent Schema Registry wire format so it interoperates with real Confluent Avro clients. |
| kafkaSchemaRegistryUrl | string | null | Confluent Schema Registry URL (Avro only). When set, the schema is registered on publish and resolved by id on consume. When omitted, MockServer runs registry-less using avroSchema and avroSchemaId. |
| avroSchema | string or object | null | Inline Avro schema (Avro only) used to encode published payloads and to decode consumed payloads in registry-less mode. May be a JSON string or an inline JSON object. |
| avroSchemaId | int | 1 | Schema id used for registry-less Avro (ignored when kafkaSchemaRegistryUrl is set). It is embedded in published messages and required on consumed messages: avroSchema describes only this one schema id, so a consumed message carrying a different id is recorded as raw text with a warning rather than decoded. Avro data carries no field names, so decoding it with the wrong schema would not fail — it would silently record the wrong values. Set this to match the id your producer writes. |
| publishOnLoad | boolean | true | Publish example messages immediately when spec is loaded. If that first publish fails (for example an AMQP message that reaches no queue), the spec still loads and the failure is reported under validationIssues in the response — so a consumer that binds its queue after MockServer starts is not locked out. |
| publishIntervalMillis | long | 0 (disabled) | Publish examples periodically at this interval. A cycle that fails is logged and the schedule continues, so publishing resumes by itself once the cause clears. |
| consume | boolean | false | Subscribe to channels and record incoming messages |
| kafkaSecurity | object | null | Security settings applied to Kafka broker connections: securityProtocol, saslMechanism, saslJaasConfig, and SSL truststore/keystore location and password |
| mqttSecurity | object | null | Security settings applied to MQTT broker connections: username, password, and an sslProperties map for TLS |
Returns the loaded spec info, active channels, publisher/subscriber counts, and recorded messages (including per-message schema validation). The Java client exposes a typed asyncApiStatus() helper; every other language issues a plain GET /mockserver/asyncapi.
MockServerClient client = new MockServerClient("localhost", 1080);
String currentStatus = client.asyncApiStatus();const response = await fetch("http://localhost:1080/mockserver/asyncapi", {
method: "GET"
});
const currentStatus = await response.text();import urllib.request
req = urllib.request.Request(
"http://localhost:1080/mockserver/asyncapi",
method="GET"
)
current_status = urllib.request.urlopen(req).read().decode("utf-8")require 'net/http'
require 'uri'
uri = URI('http://localhost:1080/mockserver/asyncapi')
current_status = Net::HTTP.get(uri)package main
import (
"io"
"net/http"
)
func main() {
resp, _ := http.Get("http://localhost:1080/mockserver/asyncapi")
defer resp.Body.Close()
currentStatus, _ := io.ReadAll(resp.Body)
_ = currentStatus
}using System.Net.Http;
using var httpClient = new HttpClient();
var currentStatus = await httpClient.GetStringAsync(
"http://localhost:1080/mockserver/asyncapi"
);use reqwest::blocking::Client;
let client = Client::new();
let current_status = client.get("http://localhost:1080/mockserver/asyncapi")
.send()
.unwrap()
.text()
.unwrap();$ch = curl_init('http://localhost:1080/mockserver/asyncapi');
curl_setopt_array($ch, [
CURLOPT_RETURNTRANSFER => true,
]);
$currentStatus = curl_exec($ch);
curl_close($ch);curl -v -X GET "http://localhost:1080/mockserver/asyncapi"Verify that messages recorded by subscribers match given criteria. This mirrors the semantics of PUT /mockserver/verify for HTTP requests.
The request body is a JSON object with the following fields:
| Field | Type | Required | Description |
|---|---|---|---|
| channel | string | yes | The channel or topic to check for messages |
| payloadSubstring | string | no | The message payload must contain this substring |
| payloadJsonPath | string | no | Dot-notation JSON path to extract from the payload (e.g. user.name) |
| expectedValue | string | no | Expected value at the JSON path (used together with payloadJsonPath) |
| count | object | no | Count constraints: {atLeast, atMost, exactly}. Default: {atLeast: 1} |
| Status | Meaning |
|---|---|
| 202 Accepted | Verification passed |
| 406 Not Acceptable | Verification failed (body contains a human-readable failure reason) |
| 400 Bad Request | Malformed request (missing channel, invalid JSON) |
| 501 Not Implemented | The mockserver-async module is not on the classpath |
The Java client exposes a typed verifyAsyncMessage(...) helper that throws an AssertionError when verification fails; every other language sends the same JSON over raw HTTP to PUT /mockserver/asyncapi/verify and checks for a 202 Accepted response.
MockServerClient client = new MockServerClient("localhost", 1080);
// throws AssertionError if verification fails
client.verifyAsyncMessage("{\"channel\":\"orders\", \"payloadJsonPath\":\"user.name\", \"expectedValue\":\"Alice\", \"count\":{\"atLeast\":1}}");const body = {
"channel": "orders",
"payloadJsonPath": "user.name",
"expectedValue": "Alice",
"count": { "atLeast": 1 }
};
const response = await fetch("http://localhost:1080/mockserver/asyncapi/verify", {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body)
});
// 202 Accepted = verification passed
console.log(response.status);import json
import urllib.request
body = {
"channel": "orders",
"payloadJsonPath": "user.name",
"expectedValue": "Alice",
"count": {"atLeast": 1}
}
req = urllib.request.Request(
"http://localhost:1080/mockserver/asyncapi/verify",
data=json.dumps(body).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="PUT"
)
# 202 Accepted = verification passed
response = urllib.request.urlopen(req)
print(response.status)require 'net/http'
require 'json'
require 'uri'
body = {
"channel" => "orders",
"payloadJsonPath" => "user.name",
"expectedValue" => "Alice",
"count" => { "atLeast" => 1 }
}
uri = URI('http://localhost:1080/mockserver/asyncapi/verify')
request = Net::HTTP::Put.new(uri, 'Content-Type' => 'application/json')
request.body = body.to_json
response = Net::HTTP.start(uri.hostname, uri.port) { |http| http.request(request) }
# 202 Accepted = verification passed
puts response.codepackage main
import (
"bytes"
"net/http"
)
func main() {
body := []byte(`{
"channel": "orders",
"payloadJsonPath": "user.name",
"expectedValue": "Alice",
"count": { "atLeast": 1 }
}`)
req, _ := http.NewRequest("PUT",
"http://localhost:1080/mockserver/asyncapi/verify",
bytes.NewReader(body))
req.Header.Set("Content-Type", "application/json")
// 202 Accepted = verification passed
http.DefaultClient.Do(req)
}using System.Net.Http;
using System.Text;
using var httpClient = new HttpClient();
var json = @"{
""channel"": ""orders"",
""payloadJsonPath"": ""user.name"",
""expectedValue"": ""Alice"",
""count"": { ""atLeast"": 1 }
}";
// 202 Accepted = verification passed
var response = await httpClient.PutAsync(
"http://localhost:1080/mockserver/asyncapi/verify",
new StringContent(json, Encoding.UTF8, "application/json")
);use reqwest::blocking::Client;
let client = Client::new();
let body = r#"{
"channel": "orders",
"payloadJsonPath": "user.name",
"expectedValue": "Alice",
"count": { "atLeast": 1 }
}"#;
// 202 Accepted = verification passed
client.put("http://localhost:1080/mockserver/asyncapi/verify")
.header("Content-Type", "application/json")
.body(body.to_string())
.send()
.unwrap();$json = <<<'JSON'
{
"channel": "orders",
"payloadJsonPath": "user.name",
"expectedValue": "Alice",
"count": { "atLeast": 1 }
}
JSON;
$ch = curl_init('http://localhost:1080/mockserver/asyncapi/verify');
curl_setopt_array($ch, [
CURLOPT_CUSTOMREQUEST => 'PUT',
CURLOPT_POSTFIELDS => $json,
CURLOPT_HTTPHEADER => ['Content-Type: application/json; charset=utf-8'],
CURLOPT_RETURNTRANSFER => true,
]);
// 202 Accepted = verification passed
curl_exec($ch);
curl_close($ch);curl -v -X PUT "http://localhost:1080/mockserver/asyncapi/verify" \
-H "Content-Type: application/json" \
-d '{
"channel": "orders",
"payloadJsonPath": "user.name",
"expectedValue": "Alice",
"count": { "atLeast": 1 }
}'
# 202 Accepted = verification passedMockServerClient client = new MockServerClient("localhost", 1080);
// throws AssertionError if any message was recorded on the "errors" channel
client.verifyAsyncMessage("{\"channel\":\"errors\", \"count\":{\"exactly\":0}}");const body = {
"channel": "errors",
"count": { "exactly": 0 }
};
const response = await fetch("http://localhost:1080/mockserver/asyncapi/verify", {
method: "PUT",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(body)
});
// 202 Accepted = verification passed
console.log(response.status);import json
import urllib.request
body = {
"channel": "errors",
"count": {"exactly": 0}
}
req = urllib.request.Request(
"http://localhost:1080/mockserver/asyncapi/verify",
data=json.dumps(body).encode("utf-8"),
headers={"Content-Type": "application/json"},
method="PUT"
)
# 202 Accepted = verification passed
response = urllib.request.urlopen(req)
print(response.status)require 'net/http'
require 'json'
require 'uri'
body = {
"channel" => "errors",
"count" => { "exactly" => 0 }
}
uri = URI('http://localhost:1080/mockserver/asyncapi/verify')
request = Net::HTTP::Put.new(uri, 'Content-Type' => 'application/json')
request.body = body.to_json
response = Net::HTTP.start(uri.hostname, uri.port) { |http| http.request(request) }
# 202 Accepted = verification passed
puts response.codepackage main
import (
"bytes"
"net/http"
)
func main() {
body := []byte(`{
"channel": "errors",
"count": { "exactly": 0 }
}`)
req, _ := http.NewRequest("PUT",
"http://localhost:1080/mockserver/asyncapi/verify",
bytes.NewReader(body))
req.Header.Set("Content-Type", "application/json")
// 202 Accepted = verification passed
http.DefaultClient.Do(req)
}using System.Net.Http;
using System.Text;
using var httpClient = new HttpClient();
var json = @"{
""channel"": ""errors"",
""count"": { ""exactly"": 0 }
}";
// 202 Accepted = verification passed
var response = await httpClient.PutAsync(
"http://localhost:1080/mockserver/asyncapi/verify",
new StringContent(json, Encoding.UTF8, "application/json")
);use reqwest::blocking::Client;
let client = Client::new();
let body = r#"{
"channel": "errors",
"count": { "exactly": 0 }
}"#;
// 202 Accepted = verification passed
client.put("http://localhost:1080/mockserver/asyncapi/verify")
.header("Content-Type", "application/json")
.body(body.to_string())
.send()
.unwrap();$json = <<<'JSON'
{
"channel": "errors",
"count": { "exactly": 0 }
}
JSON;
$ch = curl_init('http://localhost:1080/mockserver/asyncapi/verify');
curl_setopt_array($ch, [
CURLOPT_CUSTOMREQUEST => 'PUT',
CURLOPT_POSTFIELDS => $json,
CURLOPT_HTTPHEADER => ['Content-Type: application/json; charset=utf-8'],
CURLOPT_RETURNTRANSFER => true,
]);
// 202 Accepted = verification passed
curl_exec($ch);
curl_close($ch);curl -v -X PUT "http://localhost:1080/mockserver/asyncapi/verify" \
-H "Content-Type: application/json" \
-d '{
"channel": "errors",
"count": { "exactly": 0 }
}'
# 202 Accepted = verification passedAll async mocking state (publishers, subscribers, and recorded messages) is cleared when you call PUT /mockserver/reset.
| Version | Channel structure | Example resolution |
|---|---|---|
| AsyncAPI 2.x | channels.<name>.publish|subscribe.message.payload | Inline payload.example or message.examples[].payload |
| AsyncAPI 3.x | channels.<name>.messages.<msgName>.payload | examples[].payload; basic $ref to #/components/messages/<name> |
Both JSON and YAML spec formats are accepted. Missing or incomplete structures are tolerated gracefully.
When a channel's message definition includes a JSON Schema (payload), MockServer validates:
Schema validation supports JSON Schema Draft 4 through Draft 2019-09, including constraints like required, enum, minimum/maximum, pattern, and format.
Add the mockserver-async dependency to your project, then wire the components together:
import org.mockserver.async.AsyncApiMockOrchestrator;
import org.mockserver.async.asyncapi.AsyncApiParser;
import org.mockserver.async.asyncapi.AsyncApiSpec;
import org.mockserver.async.publish.KafkaMessagePublisher;
// 1. Parse your AsyncAPI spec (from a string, file, or resource)
String specYaml = Files.readString(Path.of("asyncapi.yaml"));
AsyncApiSpec spec = new AsyncApiParser().parse(specYaml);
// 2. Create a publisher pointed at your test broker
KafkaMessagePublisher publisher = new KafkaMessagePublisher("localhost:9092");
// 3. Create the orchestrator and publish once
AsyncApiMockOrchestrator orchestrator = new AsyncApiMockOrchestrator(spec, publisher);
orchestrator.publishAll();
// Or publish repeatedly on a schedule (e.g. every 500 ms)
orchestrator.startPublishing(500);
// ... run your consumer tests ...
orchestrator.stop();
// Always close the publisher to release broker connections
publisher.close();For subscribing to record messages:
import org.mockserver.async.subscribe.KafkaMessageSubscriber;
import org.mockserver.async.subscribe.RecordedMessage;
KafkaMessageSubscriber subscriber = new KafkaMessageSubscriber("localhost:9092", "test-group");
subscriber.subscribe("orders");
// ... wait for messages ...
List<RecordedMessage> messages = subscriber.getRecordedMessages("orders");
for (RecordedMessage msg : messages) {
System.out.println("Key: " + msg.getKey() + ", Payload: " + msg.getPayload());
}
subscriber.close();| Broker | Publisher | Subscriber | Features |
|---|---|---|---|
| Kafka | KafkaMessagePublisher | KafkaMessageSubscriber | Record keys, headers, consumer group |
| MQTT | MqttMessagePublisher | MqttMessageSubscriber | QoS 0/1/2, binary payloads |
| AMQP / RabbitMQ | AmqpMessagePublisher | publish-only — consumer/subscriber mocking is deferred | Exchange and queue bindings (exchange.name/type/durable, queue.name/durable, explicit routingKey); idempotent exchange declaration |
These properties provide server-wide defaults for async messaging. Per-request brokerConfig values override them.
| Property | Env Variable | Type | Default | Description |
|---|---|---|---|---|
| mockserver.asyncKafkaBootstrapServers | MOCKSERVER_ASYNC_KAFKA_BOOTSTRAP_SERVERS | string | "" (unset) | Default Kafka bootstrap servers used when the per-request brokerConfig does not include kafkaBootstrapServers. |
| mockserver.asyncMqttBrokerUrl | MOCKSERVER_ASYNC_MQTT_BROKER_URL | string | "" (unset) | Default MQTT broker URL used when the per-request brokerConfig does not include mqttBrokerUrl. |
| mockserver.asyncAmqpUri | MOCKSERVER_ASYNC_AMQP_URI | string | "" (unset) | Default AMQP/RabbitMQ connection URI used when the per-request brokerConfig does not include amqpUri. Publish-only. |
| mockserver.asyncRecordedMessageMaxEntries | MOCKSERVER_ASYNC_RECORDED_MESSAGE_MAX_ENTRIES | int | 1000 | Maximum number of recorded messages retained per channel. When the cap is reached, the oldest messages are evicted (FIFO). |
See the Configuration Properties page for the full four-form reference (Java code, system property, environment variable, property file).
The MockServerClient class provides three convenience methods that wrap the AsyncAPI control-plane endpoints:
See the per-language tabs under REST control-plane above for the Java helper alongside the equivalent raw-HTTP call in JavaScript, Python, Ruby, Go, .NET, Rust, PHP, and curl.
The AsyncAPI broker state is now visible in the MockServer dashboard's AsyncAPI (Async) view, reachable from the dashboard's top toolbar. It shows the loaded spec's channels, a publisher/subscriber summary, and messages recorded from broker subscriptions.