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.

 

See it in action

The fastest way to try AsyncAPI mocking: supply a spec and a Kafka bootstrap address and MockServer publishes example messages immediately.

  1. Start Kafka (for example with Docker Compose — a minimal single-node Kafka is sufficient).
  2. Load your AsyncAPI spec and tell MockServer to publish example messages on load and record everything that arrives on each channel:
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
    }
  }'
  1. Verify a message was published to the orders channel:
curl -s -X PUT http://localhost:1080/mockserver/asyncapi/verify \
  -H "Content-Type: application/json" \
  -d '{
    "channel": "orders",
    "count": { "atLeast": 1 }
  }'
# 202 Accepted = verification passed

Note: 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.

 

How it works

  1. Parse — AsyncApiParser reads the AsyncAPI document (JSON or YAML, auto-detected) and builds an in-memory model of channels and their message schemas.
  2. Generate — MessageExampleGenerator produces a JSON example payload for each channel. It uses explicit examples from the spec when present; otherwise it synthesises a schema-aware example (respecting enum, default, format, minimum/maximum, minLength, and other JSON Schema constraints).
  3. Publish — AsyncApiMockOrchestrator sends each payload to the broker via a MessagePublisher adapter. Kafka publishing supports record keys and headers; MQTT supports configurable QoS (0/1/2) and binary payloads.
  4. Subscribe (optional) — MessageSubscriber implementations subscribe to Kafka, MQTT, and AMQP/RabbitMQ channels and record incoming messages, including keys, headers, and schema validation results. Avro messages are decoded back to JSON before recording, so verification works the same way as for plain-JSON channels.
 

REST control-plane

Load an AsyncAPI spec and start mocking via the REST API:

Load a spec: PUT /mockserver/asyncapi

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
    }
  }'

Broker configuration options

FieldTypeDefaultDescription
kafkaBootstrapServersstringnullKafka bootstrap servers (e.g. localhost:9092)
kafkaGroupIdstringmockserver-async-consumerConsumer group ID for Kafka subscribers
mqttBrokerUrlstringnullMQTT broker URL (e.g. tcp://localhost:1883)
amqpUristringnullAMQP/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.
mqttClientIdstringmockserver-mqtt-pub / mockserver-mqtt-subMQTT 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.
mqttQosint1MQTT QoS level (0, 1, or 2)
mqttProtocolVersionint3MQTT 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.
kafkaValueFormatstringjsonKafka 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.
kafkaSchemaRegistryUrlstringnullConfluent 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.
avroSchemastring or objectnullInline 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.
avroSchemaIdint1Schema id embedded in published Avro messages in registry-less mode (ignored when kafkaSchemaRegistryUrl is set).
publishOnLoadbooleantruePublish example messages immediately when spec is loaded
publishIntervalMillislong0 (disabled)Publish examples periodically at this interval
consumebooleanfalseSubscribe to channels and record incoming messages
kafkaSecurityobjectnullSecurity settings applied to Kafka broker connections: securityProtocol, saslMechanism, saslJaasConfig, and SSL truststore/keystore location and password
mqttSecurityobjectnullSecurity settings applied to MQTT broker connections: username, password, and an sslProperties map for TLS

Check status: GET /mockserver/asyncapi

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 recorded messages: PUT /mockserver/asyncapi/verify

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:

FieldTypeRequiredDescription
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}

Responses

StatusMeaning
202 AcceptedVerification passed
406 Not AcceptableVerification failed (body contains a human-readable failure reason)
400 Bad RequestMalformed request (missing channel, invalid JSON)
501 Not ImplementedThe 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.

Example: verify a message with a specific field value

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.code
package 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 passed

Example: verify no error messages were published (negative assertion)

MockServerClient 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.code
package 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 passed

Reset

All async mocking state (publishers, subscribers, and recorded messages) is cleared when you call PUT /mockserver/reset.

 

Supported AsyncAPI versions

VersionChannel structureExample 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.

 

Schema validation

When a channel's message definition includes a JSON Schema (payload), MockServer validates:

  • Generated examples before publishing — warnings are logged and reported if a generated example does not conform to the schema
  • Recorded messages from subscriptions — each recorded message in the status response includes a schemaValid field and any schemaErrors

Schema validation supports JSON Schema Draft 4 through Draft 2019-09, including constraints like required, enum, minimum/maximum, pattern, and format.

 

Java usage

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();
 

Supported brokers

BrokerPublisherSubscriberFeatures
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
 

Configuration properties

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).

 

Java client helpers

The MockServerClient class provides three convenience methods that wrap the AsyncAPI control-plane endpoints:

  • loadAsyncApi(...) — load an AsyncAPI spec and start mocking (PUT /mockserver/asyncapi)
  • asyncApiStatus() — check the current async mocking status (GET /mockserver/asyncapi)
  • verifyAsyncMessage(...) — verify recorded messages, throwing an AssertionError on failure (PUT /mockserver/asyncapi/verify)

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.

Not yet supported

  • Advanced AsyncAPI bindings — some channel-specific binding configurations are not yet supported (e.g. Kafka partition assignment, Kafka topic-config bindings). Note: MQTT QoS, MQTT retain, and Kafka message key bindings are parsed and applied — see the Also supported section below
  • Security schemes — security scheme definitions declared in the AsyncAPI spec document are not auto-applied to broker connections. Broker security itself is supported: configure it explicitly via the kafkaSecurity and mqttSecurity broker-configuration options above (SASL, TLS truststore/keystore, username/password)
 

Also supported

  • Multi-message channels — all message definitions in a channel are published, not just the first. AsyncAPI 3.x channels with multiple messages and AsyncAPI 2.x oneOf message variants each result in one publish call per message.
  • Correlation IDs — AsyncAPI correlationId definitions (inline or via $ref to #/components/correlationIds/<name>) are parsed. At publish time MockServer generates a unique correlation ID and injects it at the message's location — either as a message header ($message.header#/…) or into the payload at the given JSON Pointer ($message.payload#/…). Note: header-location correlation IDs are delivered over Kafka headers but not over MQTT (which has no per-message headers); payload-location correlation IDs are injected for both brokers.