Async Messaging Module (mockserver-async)

Overview

The mockserver-async module provides AsyncAPI-driven message-broker mocking for Kafka, MQTT, and AMQP 0.9.1 (RabbitMQ). Given an AsyncAPI 2.x or 3.x specification document, it parses the channels and message definitions, generates schema-validated example payloads, publishes them to a message broker, and can subscribe to channels to record incoming messages for verification.

Publish + subscribe on all three brokers: Kafka, MQTT, and AMQP/RabbitMQ each support both publishing example messages and subscribing to record incoming messages for verification.

Kafka value formats: JSON (default) and Avro in the Confluent Schema Registry wire format (magic byte + schema id + Avro binary), with a registry-backed mode (schemas registered/resolved against a configured Schema Registry URL) and a registry-less mode (fixed schema id + inline schema). Protobuf is deferred.

MQTT protocol versions: MQTT 3.1.1 (default) and MQTT 5 (mqttProtocolVersion: 5), the latter adding user-property (header) delivery that v3 cannot carry.

Architecture

flowchart LR
    A["AsyncAPI Spec\n(JSON / YAML)"] --> B["AsyncApiParser"]
    B --> C["AsyncApiSpec\n(channels + messages)"]
    C --> D["MessageExampleGenerator"]
    D --> E["Example payloads\n(schema-aware)"]
    E --> F["AsyncApiMockOrchestrator"]
    F --> G["MessagePublisher"]
    G --> H["Kafka / MQTT / AMQP broker"]
    H --> I["MessageSubscriber"]
    I --> J["RecordedMessages"]

Control-Plane Integration

flowchart TD
    Client["HTTP Client"] -->|PUT /mockserver/asyncapi| HS["HttpState\n(mockserver-core)"]
    Client -->|GET /mockserver/asyncapi| HS
    Client -->|PUT /mockserver/asyncapi/verify| HS
    HS --> REG["AsyncApiControlPlaneRegistry\n(SPI holder in core)"]
    REG --> IMPL["AsyncApiControlPlaneImpl\n(mockserver-async)"]
    IMPL --> Parser["AsyncApiParser"]
    IMPL --> Pub["Publishers"]
    IMPL --> Sub["Subscribers"]
    IMPL --> Val["Schema Validator"]

The control-plane uses an SPI/registry pattern (Option A from the design spec): a lightweight AsyncApiControlPlane interface and AsyncApiControlPlaneRegistry holder live in mockserver-core, keeping core free of async module dependencies. The actual implementation (AsyncApiControlPlaneImpl) lives in mockserver-async and self-registers at server startup via reflection from MockServer.createServerBootstrap(). This mirrors the pattern established by GrpcHealthRegistry, WasmStore, DriftStore, and other optional subsystem registries in core.

Key Classes

Class	Package	Responsibility
AsyncApiControlPlane	o.m.async (core)	SPI interface for the control-plane
AsyncApiControlPlaneRegistry	o.m.async (core)	Singleton holder; routes HttpState calls to the implementation
AsyncApiControlPlaneImpl	o.m.async.controlplane	Full implementation: load, status, reset, verify, broker lifecycle
AsyncApiParser	o.m.async.asyncapi	Parses AsyncAPI 2.x/3.x JSON or YAML into an AsyncApiSpec model
AsyncApiSpec	o.m.async.asyncapi	Immutable model: version, title, list of AsyncApiChannel
AsyncApiChannel	o.m.async.asyncapi	A channel name, payload examples, optional JSON Schema, parsed bindings (MQTT qos/retain, Kafka key), and optional multi-message list
AsyncApiMessage	o.m.async.asyncapi	A single message definition: name, payload schema, payload examples, Kafka key binding, correlation ID location
MessageExampleGenerator	o.m.async	Schema-aware example generation (enum, default, format, min/max, minLength, const); per-channel and per-message
AsyncApiSchemaValidator	o.m.async.validation	Validates payloads against channel JSON Schemas using core's JsonSchemaValidator
PublishOptions	o.m.async.publish	Immutable carrier for per-message publish-time options: Kafka key, MQTT qos, MQTT retain, message headers (e.g. correlation ID)
MessagePublisher	o.m.async.publish	Interface: publish(channel, payload), publish(channel, key, payload, headers), publish(channel, payload, options), close()
KafkaMessagePublisher	o.m.async.publish	Wraps KafkaProducer; supports keys and headers
MqttMessagePublisher	o.m.async.publish	Wraps Paho v3 MqttClient; supports configurable QoS (0/1/2) and binary payloads
Mqtt5MessagePublisher	o.m.async.publish	Wraps Paho v5 MqttClient; QoS/retain/binary as v3 plus MQTT 5 user properties (headers) from PublishOptions.getHeaders()
KafkaAvroMessagePublisher	o.m.async.publish	Wraps KafkaProducer<String,byte[]>; encodes JSON payloads to Confluent-wire-format Avro (magic byte + schema id + Avro binary); registry-backed or registry-less schema id
AmqpMessagePublisher	o.m.async.publish	Wraps the RabbitMQ com.rabbitmq.client.Channel; derives the exchange + routing key from each channel's AmqpBinding, declares the exchange/queue idempotently, and emits headers (e.g. correlation IDs) as AMQP message properties
AmqpBinding	o.m.async.asyncapi	Immutable model of the AsyncAPI AMQP channel binding (is, exchange name/type/durable, queue name/durable, routing key)
MessageSubscriber	o.m.async.subscribe	Interface: subscribe(channel), unsubscribe(channel), getRecordedMessages(), close()
KafkaMessageSubscriber	o.m.async.subscribe	Wraps KafkaConsumer with background poll loop; all consumer access confined to the poll thread via a queued-ops pattern; records messages in bounded stores
KafkaAvroMessageSubscriber	o.m.async.subscribe	Wraps KafkaConsumer<String,byte[]>; strips the Confluent wire-format header, resolves the schema (registry by id, or inline), and decodes Avro binary to JSON before recording; falls back to raw string for non-Avro bytes
MqttMessageSubscriber	o.m.async.subscribe	Wraps Paho v3 MqttClient callback; records messages in bounded stores
Mqtt5MessageSubscriber	o.m.async.subscribe	Wraps Paho v5 MqttClient callback; records MQTT 5 user properties as message headers
AmqpMessageSubscriber	o.m.async.subscribe	Consumes from RabbitMQ; resolves the queue from the channel's AmqpBinding (queue-based: the named queue; routingKey-based: declares the exchange + a private queue bound on the routing key); records message-property headers
ConfluentWireFormat	o.m.async.serde	Encodes/decodes the Confluent wire-format header (magic byte 0x00 + 4-byte big-endian schema id + payload); codec-agnostic and broker-free
AvroPayloadCodec	o.m.async.serde	Converts payloads between JSON text and Avro binary via Apache Avro GenericDatum reader/writer (no code generation, no Confluent serde stack)
SchemaRegistryClient	o.m.async.serde	Minimal Confluent Schema Registry REST client over the JDK HttpClient: getSchemaById(id) and register(subject, schema), both cached
Mqtt5SecurityOptions	o.m.async.security	Builds an MQTT v5 MqttConnectionOptions from MqttSecurity (username/password/SSL) — the v5 counterpart of MqttSecurityOptions
BoundedMessageStore	o.m.async.subscribe	Thread-safe, bounded FIFO store for RecordedMessage instances (default 1000 per channel); evicts oldest when full
RecordedMessage	o.m.async.subscribe	Immutable record: channel, key, payload, headers, timestamp
AsyncApiMockOrchestrator	o.m.async	Publishes examples once (publishAll()) or on a schedule (startPublishing(interval) / stop())

REST Control-Plane

PUT /mockserver/asyncapi

Load an AsyncAPI spec and start mocking. The request body can be either:

1. Plain spec: the AsyncAPI document as JSON or YAML
2. Wrapped body: {"spec": <spec>, "brokerConfig": {...}}

Broker configuration options (brokerConfig):

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)
mqttClientId	string	mockserver-mqtt-pub/sub	MQTT client ID prefix
mqttQos	int	1	MQTT QoS level (0, 1, or 2)
mqttProtocolVersion	int	3	MQTT protocol version: 3 (3.1.1) or 5 (adds user-property/header delivery)
amqpUri	string	null	AMQP (RabbitMQ) connection URI (e.g. amqp://guest:guest@localhost:5672/)
kafkaValueFormat	string	json	Kafka value serialization: json or avro (Confluent wire format)
kafkaSchemaRegistryUrl	string	null	Confluent Schema Registry URL (Avro only); when present, schema ids are registered on publish and resolved by id on consume
avroSchema	string | object	null	Inline Avro schema (Avro only): the schema 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	Fixed schema id embedded in published messages in registry-less mode (ignored when a registry URL is set)
publishOnLoad	boolean	true	Publish examples immediately on load
publishIntervalMillis	long	0	Schedule periodic publishing (0 = disabled)
consume	boolean	false	Enable consumer/subscriber for each channel (Kafka, MQTT, and AMQP)
kafkaSecurity	object	null	Kafka SASL/SSL security config (see Broker Security)
mqttSecurity	object	null	MQTT username/password/SSL security config (see Broker Security)

Response (201 Created):

{
  "loaded": true,
  "specTitle": "My API",
  "specVersion": "2.6.0",
  "channelCount": 2,
  "channels": [{"name": "orders", "hasSchema": true}],
  "publishers": 1,
  "subscribers": 1
}

GET /mockserver/asyncapi

Returns current status including loaded spec info, active channels, and recorded messages from subscribers.

Response (200 OK):

{
  "loaded": true,
  "specTitle": "My API",
  "specVersion": "2.6.0",
  "channels": [{"name": "orders", "hasSchema": true, "exampleCount": 1}],
  "publishers": 1,
  "subscribers": 1,
  "recordedMessages": [
    {
      "channel": "orders",
      "key": "order-123",
      "payload": "{\"orderId\":42}",
      "headers": {"trace-id": "abc"},
      "timestamp": "2024-01-01T00:00:00Z",
      "schemaValid": true
    }
  ]
}

PUT /mockserver/asyncapi/verify

Verify that recorded messages match the given criteria. Mirrors the semantics of PUT /mockserver/verify for HTTP requests.

Request body (JSON):

Field	Type	Required	Description
channel	string	yes	The channel/topic to check
payloadSubstring	string	no	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 with payloadJsonPath)
count	object	no	Count constraints: {atLeast, atMost, exactly}. Default: {atLeast: 1}

Responses:

Status	Meaning
202 Accepted	Verification passed
406 Not Acceptable	Verification failed (body contains human-readable failure reason)
400 Bad Request	Malformed request (missing channel, invalid JSON)
501 Not Implemented	mockserver-async module is not on the classpath

Example — verify at least 1 message on "orders" with a specific user name:

{
  "channel": "orders",
  "payloadJsonPath": "user.name",
  "expectedValue": "Alice",
  "count": { "atLeast": 1 }
}

Example — verify exactly 0 messages on a channel (negative assertion):

{
  "channel": "errors",
  "count": { "exactly": 0 }
}

PUT /mockserver/asyncapi/http

Import an AsyncAPI spec as HTTP mock expectations instead of loading it into the broker-publishing mock. One GET expectation is created per channel, returning that channel's schema-aware example payload (the same payload MessageExampleGenerator produces for broker publishing). This lets a consumer poll example messages over plain HTTP with no live Kafka/MQTT/AMQP broker.

It reuses the same AsyncApiParser and MessageExampleGenerator as the broker path (in AsyncApiHttpExpectationGenerator); the generated expectations are serialized with core's ExpectationSerializer and returned across the SPI as a JSON array so HttpState can deserialize and add them without a compile-time dependency on the parser.

Request body: a plain AsyncAPI spec (JSON/YAML), or a wrapper {"spec": "...", "channelPathPrefix": "/events"}. Channel names are mapped to URL paths (dots become slashes; collisions get a numeric suffix); channelPathPrefix is prepended when supplied.

Responses:

Status	Meaning
201 Created	One GET expectation created per channel (body is the upserted expectation array)
400 Bad Request	Missing or unparseable spec, or a spec with no channels
501 Not Implemented	mockserver-async module is not on the classpath

Reset

All async mocking state (publishers, subscribers, recorded messages) is cleared on PUT /mockserver/reset.

AsyncAPI Parsing

The parser auto-detects JSON vs YAML (by leading { character) and supports:

• AsyncAPI 2.x: channels.<name>.publish|subscribe.message.payload for schema; .payload.example for inline examples; .message.examples[].payload for the examples array; message.oneOf for multi-message channels (each variant becomes a separate AsyncApiMessage)
• AsyncAPI 3.x: channels.<name>.messages.<msgName>.payload for schema; .examples[].payload for examples; basic $ref resolution to #/components/messages/<name>; all messages under channels.<name>.messages are parsed (not just the first)

Missing or incomplete structures are tolerated gracefully (channels appear with empty examples).

Multi-Message Channels

A channel may define multiple message types:

• AsyncAPI 3.x: multiple entries under channels.<name>.messages (e.g. userCreated, orderPlaced)
• AsyncAPI 2.x: message.oneOf array in the operation (e.g. publish.message.oneOf: [{...}, {...}])

When a channel has multiple messages:

1. Each message is parsed into an AsyncApiMessage with its own payload schema, examples, and Kafka key binding
2. The channel's getMessages() method returns the full list of AsyncApiMessage instances
3. The orchestrator publishes one example per message (not just one per channel)
4. Per-message PublishOptions combine the message's Kafka key with the channel-level MQTT qos/retain
5. Legacy single-message accessors (getPayloadExamples(), getPayloadSchema(), getKafkaKey()) continue to return the first message's values for backward compatibility

For single-message channels, getMessages() synthesizes a single-element list from the channel's existing fields, so all callers can use the uniform per-message API without conditional logic.

Example Generation (Schema-Aware)

The MessageExampleGenerator follows this precedence per channel:

1. First explicit example from the spec
2. Schema-aware synthesis from JSON Schema, respecting:
   • default values
   • enum (uses first value)
   • const values
   • minimum/maximum and exclusiveMinimum/exclusiveMaximum
   • minLength (pads string to required length)
   • minItems (pads array to required size)
   • format (generates format-appropriate values: date-time, email, uuid, uri, ipv4, ipv6)
   • pattern (heuristic matching for common patterns like email, numeric)
3. Fallback: {}

Schema Validation

The AsyncApiSchemaValidator reuses core's JsonSchemaValidator (backed by com.networknt:json-schema-validator) to validate:

• Generated examples before publishing (warnings logged for non-conforming examples)
• First spec-provided message example per channel at load time (see below)
• Consumed/recorded messages from broker subscriptions (validation result included in status response)

First-Message Example Validation (Load-Time)

When an AsyncAPI spec is loaded via PUT /mockserver/asyncapi, the control plane validates the first payload example of each message in each channel against that message's declared payload schema. This catches malformed spec examples early, before any broker interaction.

• Scope: only the first example per message is validated (not all examples).
• Skip conditions: messages with no payload schema or no explicit examples are skipped cleanly.
• Multi-message channels (v3 multiple messages, v2 oneOf): each message's first example is validated independently.
• Surfacing: non-conforming examples are surfaced as:
  • A WARN-level log line (matching the existing generated-example warning pattern)
  • A validationIssues entry in the load response and status response, with context first_message_example (or first_message_example:<messageName> when the message has a name)
• No hard failure: the load succeeds even when examples fail validation, matching the existing convention for generated-example validation.
• No publish-time impact: this validation is purely at load time and does not affect publish-time behaviour.

Consumer/Subscriber Mocking

MockServer can subscribe to Kafka topics, MQTT topics, and AMQP queues/exchanges to record incoming messages, mirroring how HTTP requests are recorded for verification. Subscribers are created when consume: true is set in the broker config.

AMQP subscribe: the queue to consume from is derived from the channel's AmqpBinding, mirroring the publisher's destination resolution: a queue-based channel (or a channel with no binding) consumes the named queue directly; a routingKey-based channel declares the exchange, declares a private (server-named, exclusive, auto-delete) queue, and binds it to the exchange on the routing key. AMQP message-property headers are recorded as the RecordedMessage headers.

Recorded messages include:

• Channel/topic name
• Message key (Kafka) or null (MQTT)
• Payload (string)
• Headers (Kafka) or empty map (MQTT)
• Timestamp
• Schema validation result (when a schema is defined)

Recorded messages are stored in a bounded BoundedMessageStore per channel (default 1000 messages). When the cap is reached, the oldest message is evicted (FIFO). This prevents unbounded memory growth under high message volume.

Message Keys, QoS, and Headers

• Kafka: KafkaMessagePublisher.publish(channel, key, payload, headers) supports configurable record keys and arbitrary headers
• MQTT: MqttMessagePublisher supports configurable QoS (0, 1, or 2) and binary payloads via publishBytes()
• Kafka Consumer: KafkaMessageSubscriber records message keys and headers from consumed records

Kafka Avro / Confluent Schema Registry

Set kafkaValueFormat: "avro" to have MockServer publish and consume Kafka messages in the Confluent Schema Registry wire format ([0x00][schemaId:4][Avro binary]) so real Confluent Avro producers/consumers interoperate with the mock.

flowchart LR
    J["JSON example\n(from spec)"] --> C["AvroPayloadCodec\njsonToAvro"]
    C --> W["ConfluentWireFormat\nencode(schemaId, avro)"]
    W --> K["Kafka topic\n(byte[] value)"]
    K --> D["ConfluentWireFormat\ndecode"]
    D --> R["resolve schema\n(registry by id / inline)"]
    R --> A["AvroPayloadCodec\navroToJson"]
    A --> M["RecordedMessage\n(JSON payload)"]

Design decision — Apache Avro, not the Confluent serde stack. The Confluent kafka-avro-serializer/kafka-schema-registry-client artifacts are under the Confluent Community License and pull in a heavy dependency tree. MockServer instead uses Apache Avro (Apache 2.0) for the JSON<->binary codec (AvroPayloadCodec), a hand-rolled 5-byte framing (ConfluentWireFormat), and a minimal JDK-HttpClient Schema Registry REST client (SchemaRegistryClient). This keeps licensing clean and dependencies small while remaining byte-compatible with Confluent producers/consumers.

Two modes:

Mode	Trigger	Publish schema id	Consume schema resolution
Registry-backed	kafkaSchemaRegistryUrl set	schema registered under <topic>-value, returned id embedded	fetched by embedded id via GET /schemas/ids/{id} (cached)
Registry-less	no kafkaSchemaRegistryUrl	fixed avroSchemaId (default 1) embedded	the inline avroSchema (id ignored)

Both modes require an avroSchema (inline) for the publish path (encoding needs a schema) and for registry-less consume (decoding without a registry needs the inline schema). Consumed bytes that are not wire-format-framed, or that fail to decode, are recorded as their raw UTF-8 string rather than dropped. Because the recorded payload is JSON, the existing PUT /mockserver/asyncapi/verify substring / JSON-path checks work unchanged against Avro messages.

JSON encoding note: Avro's JSON representation is stricter than plain JSON for union-typed (nullable/optional) fields, which need the {"fieldType": value} form. Flat records of primitive fields — the common case for AsyncAPI example payloads — map straight across.

MQTT Protocol Version (3.1.1 and 5)

mqttProtocolVersion selects the Paho client: 3 (default, MqttMessagePublisher/MqttMessageSubscriber, Paho mqttv3) or 5 (Mqtt5MessagePublisher/Mqtt5MessageSubscriber, Paho mqttv5). QoS, retain, and binary payloads behave identically across versions. The v5 advantage is user properties: PublishOptions.getHeaders() (e.g. header-location correlation IDs) are delivered as MQTT 5 user properties on publish and recorded as RecordedMessage headers on consume — neither of which MQTT 3 can carry. Security (mqttSecurity: username/password/SSL) applies to both versions via MqttSecurityOptions (v3) / Mqtt5SecurityOptions (v5).

Paho v5 subscribe caveat: Paho mqttv5 1.2.5's MqttClient.subscribe(topic, qos, listener) overload self-recurses (StackOverflow). Mqtt5MessageSubscriber deliberately uses the safe 2-arg subscribe(topic, qos) together with a setCallback(...) message callback, and the live tests observe messages through that subscriber rather than a per-subscription listener.

AsyncAPI Channel Bindings

The parser extracts publish-time bindings from the AsyncAPI spec and threads them through to the publishers via an immutable PublishOptions carrier. The following bindings are supported:

Supported Bindings

Binding	AsyncAPI Location	Applies To	Effect
MQTT QoS	publish.bindings.mqtt.qos (v2), channels.<n>.bindings.mqtt.qos (v3 best-effort)	MqttMessagePublisher	Overrides the instance-level QoS for the message
MQTT retain	publish.bindings.mqtt.retain (v2), channels.<n>.bindings.mqtt.retain (v3 best-effort)	MqttMessagePublisher	Sets MqttMessage.setRetained()
Kafka message key	publish.message.bindings.kafka.key (v2), messages.<n>.bindings.kafka.key (v3)	KafkaMessagePublisher	Sets the ProducerRecord key

AMQP Channel Bindings

For AMQP, the destination is not the channel name directly — it is derived from the channel's bindings.amqp definition. The parser reads channels.<name>.bindings.amqp (channel-level in both v2 and v3) into an AmqpBinding, and AmqpMessagePublisher resolves it to an (exchange, routing-key) pair:

is value	Exchange	Routing key	Notes
routingKey (default)	exchange.name (or the default exchange "" when absent)	the binding's routingKey, else the channel name	Exchange is declared idempotently using exchange.type (default direct) and exchange.durable (default true) before the first publish
queue	default exchange ("")	queue.name (or the channel name)	The named queue is declared (durable per queue.durable) so the message is not dropped against a fresh broker

When a channel has no bindings.amqp, the publisher falls back to the default exchange with the channel name as the routing key — mirroring the topic-name-as-destination convention of the Kafka/MQTT publishers.

Supported binding fields: is, exchange.name, exchange.type, exchange.durable, queue.name, queue.durable, and an explicit routingKey extension. Correlation-ID/PublishOptions headers are emitted as AMQP message properties.

Deferred (not applied at publish time): exchange.autoDelete, exchange.vhost, queue.exclusive, queue.autoDelete, queue.vhost, and operation/message-level AMQP bindings (cc, bcc, deliveryMode, mandatory, replyTo, priority, timestamp, expiration).

Kafka Key Extraction

The Kafka key binding (bindings.kafka.key) can be:

• A scalar literal (string or number) -- used directly
• A schema with const -- the const value is used
• A schema with example -- the example value is used
• A schema with examples[] -- the first example is used
• A bare schema (no literal derivable) -- key is null (not applied)

PublishOptions Threading

The AsyncApiMockOrchestrator iterates each channel's messages via channel.getMessages(), generates an example for each message, and builds per-message PublishOptions combining the message's Kafka key with the channel-level MQTT qos/retain. Each message results in a separate publisher.publish(channel, payload, options) call. For single-message channels, this is equivalent to the previous one-publish-per-channel behavior. Publishers that do not override the publish(channel, payload, options) method fall back to the default implementation which ignores the options, preserving backward compatibility.

Correlation IDs

AsyncAPI messages may define a correlationId with a location runtime expression that specifies where a correlation identifier should be placed in the published message. The orchestrator generates a unique correlation ID (UUID by default) at publish time and injects it at the specified location.

Supported Location Expressions

Expression Pattern	Example	Effect
$message.header#/<headerName>	$message.header#/correlationId	Adds a header {headerName: id} to PublishOptions.headers; Kafka publishers emit it as a RecordHeader
$message.payload#/<jsonPointer>	$message.payload#/metadata/id	Injects the correlation ID into the JSON payload at the given JSON Pointer path (creates intermediate objects if needed)

Unrecognised location prefixes are skipped with a DEBUG log (never throw).

Parsing

The parser reads message.correlationId from each message definition (v2 single, v2 oneOf variant, v3 per-message). If it is a $ref (e.g. $ref: "#/components/correlationIds/defaultId"), it is resolved using the same resolveRef mechanism used for message references. The location string is stored as AsyncApiMessage.correlationIdLocation.

Injection at Publish Time

In AsyncApiMockOrchestrator.publishAll(), for each message with a non-null correlationIdLocation:

1. A unique correlation ID is generated via an injectable Supplier<String> (defaults to UUID.randomUUID().toString(); tests can pin it to a fixed value for deterministic assertions).
2. Header location ($message.header#/...): the header name is extracted from the location suffix, and a {headerName: correlationId} entry is added to the PublishOptions.headers map. KafkaMessagePublisher passes these as Kafka RecordHeader values.
3. Payload location ($message.payload#/...): the JSON Pointer is extracted, the generated example payload is parsed as JSON (Jackson ObjectMapper), the value is set at the pointer path (creating intermediate ObjectNode containers as needed), and the payload is re-serialized. If the payload is not valid JSON, injection is skipped with a DEBUG log.
4. One correlation ID is generated per message-publish. The injection happens before publisher.publish() is called.

MQTT Header Limitation

MQTT does not support message-level headers. When PublishOptions.headers is non-empty and the publisher is MqttMessagePublisher, a DEBUG log is emitted noting that header-location correlation IDs are not delivered over MQTT. Payload-location correlation IDs work with MQTT because they are injected into the payload JSON before publishing.

Limitations

• v3 MQTT operation bindings: In AsyncAPI 3.x, MQTT QoS and retain are properly located in the top-level operations section's bindings, not on channels. The parser checks for channel-level bindings.mqtt as a best-effort fallback but does not navigate v3 operation-to-channel references. Full v3 operation-binding resolution is deferred.
• Kafka topic-config bindings: Kafka channel bindings for topic configuration (partitions, replicas, cleanup policy) are intentionally not applied at publish time -- they describe topic creation parameters, not message-level settings.

Broker Security

The brokerConfig supports optional security configuration for connecting to enterprise brokers that require SASL authentication and/or TLS. When security is absent or empty, the adapters use plaintext connections (backward compatible).

Kafka Security (kafkaSecurity)

Field	Kafka Config Key	Description
securityProtocol	security.protocol	Protocol: PLAINTEXT, SSL, SASL_PLAINTEXT, SASL_SSL
saslMechanism	sasl.mechanism	SASL mechanism: PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, OAUTHBEARER
saslJaasConfig	sasl.jaas.config	JAAS login module configuration string
sslTruststoreLocation	ssl.truststore.location	Path to the SSL truststore file
sslTruststorePassword	ssl.truststore.password	Password for the SSL truststore
sslKeystoreLocation	ssl.keystore.location	Path to the SSL keystore file (for mTLS)
sslKeystorePassword	ssl.keystore.password	Password for the SSL keystore
sslKeyPassword	ssl.key.password	Password for the private key in the keystore

Example:

{
  "kafkaBootstrapServers": "broker:9093",
  "kafkaSecurity": {
    "securityProtocol": "SASL_SSL",
    "saslMechanism": "PLAIN",
    "saslJaasConfig": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"user\" password=\"pass\";",
    "sslTruststoreLocation": "/path/to/truststore.jks",
    "sslTruststorePassword": "changeit"
  }
}

MQTT Security (mqttSecurity)

Field	Description
username	MQTT broker username
password	MQTT broker password
sslProperties	Map of SSL properties passed to Paho MqttConnectOptions.setSSLProperties()

Supported SSL property keys (Paho/IBM conventions):

Key	Description
com.ibm.ssl.keyStore	Path to the client keystore
com.ibm.ssl.keyStorePassword	Keystore password
com.ibm.ssl.trustStore	Path to the truststore
com.ibm.ssl.trustStorePassword	Truststore password
com.ibm.ssl.protocol	SSL protocol (e.g. TLSv1.2)

Example:

{
  "mqttBrokerUrl": "ssl://broker:8883",
  "mqttSecurity": {
    "username": "user",
    "password": "pass",
    "sslProperties": {
      "com.ibm.ssl.trustStore": "/path/to/truststore.jks",
      "com.ibm.ssl.trustStorePassword": "changeit"
    }
  }
}

Implementation

Security configuration is applied through testable utility classes:

• KafkaSecurityProperties.applySecurity(Properties, KafkaSecurity) — sets the Kafka config keys on the client properties
• MqttSecurityOptions.buildConnectOptions(MqttSecurity) — builds MqttConnectOptions with username/password/SSL (returns null when empty, preserving the no-arg connect() path)

The security objects are parsed from the brokerConfig JSON in AsyncApiControlPlaneImpl.parseBrokerConfig() and threaded through to the publisher/subscriber constructors.

Metrics

When the server is started with metrics enabled (mockserver.metricsEnabled=true), the async module emits two Prometheus counters labelled by channel:

Metric	Incremented
mock_server_async_messages_published_total{channel}	Once per message published to a broker (AsyncApiMockOrchestrator.publishAll() — both publish-on-load and scheduled publishing)
mock_server_async_messages_consumed_total{channel}	Once per message recorded from a broker subscription (KafkaMessageSubscriber / MqttMessageSubscriber record path)

The module calls the static Metrics.incrementAsyncMessagePublished(channel) / Metrics.incrementAsyncMessageConsumed(channel) helpers in mockserver-core; they are null-safe no-ops when metrics are disabled, so the async hot paths pay nothing when metrics are off. The optional Maven scope on mockserver-core simply keeps it from being re-exported to downstream consumers of mockserver-async — the static import is still a compile-time dependency. The counters move only when a real broker is connected — a broker-less spec load (no kafkaBootstrapServers/mqttBrokerUrl) leaves them at zero. The dashboard Metrics view charts them on a dedicated "Async message activity" panel, separate from HTTP request activity. See metrics.md.

Build/Docker Wiring

The mockserver-async module is wired into the running server:

• mockserver-netty declares mockserver-async as an optional dependency
• mockserver-netty-no-dependencies (the standalone/Docker jar) explicitly includes mockserver-async so it's bundled by the shade plugin
• Registration: MockServer.createServerBootstrap() uses reflection to call AsyncApiControlPlaneImpl.registerIfAvailable() at startup, avoiding a hard compile-time dependency
• When the module is absent from the classpath, the /mockserver/asyncapi endpoints respond with 501 (Not Implemented)

Dependencies

Dependency	Version	Purpose
jackson-databind	(parent-managed)	JSON parsing and generation
jackson-dataformat-yaml	(parent-managed)	YAML parsing
kafka-clients	3.9.2	Kafka producer and consumer
org.eclipse.paho.client.mqttv3	1.2.5	MQTT 3.1.1 client (publish and subscribe)
org.eclipse.paho.mqttv5.client	1.2.5	MQTT 5 client (publish and subscribe)
org.apache.avro:avro	1.12.0	Avro JSON<->binary codec for the Kafka Confluent wire format (Apache 2.0 — avoids the Confluent Community License serde stack)
com.rabbitmq:amqp-client	5.28.0	AMQP 0.9.1 (RabbitMQ) client (publish and subscribe)
mockserver-core	(optional)	SPI interface, JSON Schema validator, shared utilities

Tests

Test Class	What it covers
AsyncApiParserTest	AsyncAPI 2.x/3.x parsing (JSON, YAML, refs, edge cases, binding extraction)
MessageExampleGeneratorTest	Basic example generation (explicit, synthesized, fallback)
MessageExampleGeneratorSchemaAwareTest	Schema-aware synthesis (enum, default, format, min/max, const, minLength, minItems)
PublishOptionsTest	PublishOptions construction, validation, isEmpty, qos range checking
AsyncApiMockOrchestratorTest	Orchestrator publish/schedule lifecycle, PublishOptions threading (mocked publisher)
KafkaMessagePublisherTest	Basic Kafka publishing (mocked producer)
KafkaMessagePublisherKeyHeadersTest	Kafka keys, headers, and PublishOptions (key from bindings) (mocked producer)
MqttMessagePublisherTest	Basic MQTT publishing (mocked client)
MqttMessagePublisherQosTest	MQTT QoS, binary payloads, and PublishOptions (retain, qos override) (mocked client)
BoundedMessageStoreTest	Bounded FIFO store: capacity, eviction, snapshot isolation, edge cases
KafkaMessageSubscriberTest	Kafka subscribing, message recording, bounded eviction, queued-ops pattern (mocked consumer)
MqttMessageSubscriberTest	MQTT subscribing, message recording, bounded eviction (mocked client)
AsyncApiSchemaValidatorTest	Schema validation (required, type, enum, min/max, pattern)
AsyncApiFirstMessageExampleValidationTest	Per-message first-example schema validation at load time: conforming, non-conforming, no-schema, no-example, multi-message, first-only scope, context naming
AsyncApiControlPlaneImplTest	Control-plane load/status/reset lifecycle (no real broker)
AsyncApiControlPlaneSecurityTest	Security scheme parsing from brokerConfig JSON (kafkaSecurity, mqttSecurity, edge cases)
AsyncApiControlPlaneVerifyTest	Message verification: count semantics (atLeast/atMost/exactly), payload substring, JSON path matching, error cases
AsyncApiControlPlaneRegistryTest	SPI holder delegation (including verify) and not-available responses (in core)
KafkaSecurityPropertiesTest	KafkaSecurityProperties.applySecurity(): all/partial/empty/null security, blank value skipping
KafkaMessagePublisherSecurityTest	buildProducerProperties() with SASL_SSL, null, empty, and partial security
KafkaMessageSubscriberSecurityTest	buildConsumerProperties() with SASL_SSL, SCRAM, null, empty security
MqttSecurityOptionsTest	MqttSecurityOptions.buildConnectOptions(): username/password, SSL properties, null/empty security
AmqpBindingParserTest	AsyncAPI AMQP bindings.amqp parsing (routingKey/queue, exchange/queue fields, defaults, no-binding, explicit routingKey, isolation from Kafka/MQTT)
AmqpMessagePublisherTest	AMQP destination derivation and publishing via a mocked RabbitMQ Channel (no broker): exchange/queue/routing-key resolution, idempotent declare, header properties, close
KafkaLiveBrokerIntegrationTest	Docker-gated (Testcontainers): publish/consume via real Kafka, subscriber recording, orchestrator end-to-end
MqttLiveBrokerIntegrationTest	Docker-gated (Testcontainers): publish/receive via real Mosquitto, subscriber recording, orchestrator end-to-end, round-trip
AmqpMessageSubscriberTest	AMQP subscribe with a mocked RabbitMQ Channel (no broker): queue resolution (queue-based, exchange/routing-key, no-binding), basicConsume registration, DeliverCallback recording, header extraction, unsubscribe cancel, bounded eviction
Mqtt5MessagePublisherTest	MQTT 5 publish with a mocked v5 client: QoS/retain, binary payloads, user-property (header) delivery, invalid-QoS guard
Mqtt5MessageSubscriberTest	MQTT 5 subscribe with a mocked v5 client: recording, user-properties-as-headers, disconnect health, bounded eviction
Mqtt5SecurityOptionsTest	Mqtt5SecurityOptions.buildConnectOptions(): username/password, SSL properties, null/empty security, blank-field skipping
ConfluentWireFormatTest	Wire-format framing: encode/decode round-trip, magic-byte + big-endian schema id layout, detection, empty payload, error cases
AvroPayloadCodecTest	Avro JSON<->binary round-trip, type preservation, distinct encodings, schema-mismatch error
SchemaRegistryClientTest	Schema Registry REST client with a stubbed HttpClient: get-by-id, register, caching, non-200 handling, blank-URL guard
KafkaAvroMessagePublisherTest	Confluent framing via a mocked producer: registry-less fixed id, registry id, key/headers, missing-schema and schema-mismatch errors
AsyncApiControlPlaneParityConfigTest	parseBrokerConfig of the parity fields: MQTT protocol version, Kafka Avro value format, Schema Registry URL, inline Avro schema (string or object)
AmqpLiveBrokerIntegrationTest	Docker-gated (Testcontainers): publish and subscribe/record against a real RabbitMQ broker — queue-bound channel, exchange + routing-key channel, orchestrator end-to-end from a parsed spec
Mqtt5LiveBrokerIntegrationTest	Docker-gated (Testcontainers/Mosquitto): MQTT 5 publish/receive, user-property round-trip, subscriber recording, adapter round-trip
KafkaAvroLiveBrokerIntegrationTest	Docker-gated (Testcontainers/Kafka): registry-less Confluent-framed Avro publish readable by a plain byte[] consumer, and end-to-end round-trip through the MockServer Avro publisher/subscriber

Configuration Properties (Async Defaults)

The following ConfigurationProperties provide server-wide defaults for async messaging. Values set via the request body's brokerConfig take precedence over these defaults.

Property Key	Env Var	Type	Default	Description
mockserver.asyncKafkaBootstrapServers	MOCKSERVER_ASYNC_KAFKA_BOOTSTRAP_SERVERS	String	"" (none)	Default Kafka bootstrap servers; used when brokerConfig.kafkaBootstrapServers is absent
mockserver.asyncMqttBrokerUrl	MOCKSERVER_ASYNC_MQTT_BROKER_URL	String	"" (none)	Default MQTT broker URL; used when brokerConfig.mqttBrokerUrl is absent
mockserver.asyncAmqpUri	MOCKSERVER_ASYNC_AMQP_URI	String	"" (none)	Default AMQP (RabbitMQ) connection URI; used when brokerConfig.amqpUri is absent
mockserver.asyncRecordedMessageMaxEntries	MOCKSERVER_ASYNC_RECORDED_MESSAGE_MAX_ENTRIES	int	1000	Maximum recorded messages per channel in subscriber stores

These properties follow the standard four-form pattern (system property, environment variable, property file, Configuration instance setter) used by all other MockServer configuration properties.

Java Client Helpers

MockServerClient provides fluent helpers for the asyncapi control-plane:

Method	HTTP Call	Return / Throw
loadAsyncApi(String specOrWrappedJson)	PUT /mockserver/asyncapi	JSON response string (spec info)
asyncApiStatus()	GET /mockserver/asyncapi	JSON status string
verifyAsyncMessage(String verificationJson)	PUT /mockserver/asyncapi/verify	Returns this on 202; throws AssertionError on 406 with the failure description

Example usage:

MockServerClient client = new MockServerClient("localhost", 1080);
client.loadAsyncApi("{\"spec\":{...}, \"brokerConfig\":{\"kafkaBootstrapServers\":\"localhost:9092\"}}");
String status = client.asyncApiStatus();
client.verifyAsyncMessage("{\"channel\":\"orders\",\"count\":{\"atLeast\":1}}");

Deferred (Honest List)

The following items are not yet implemented:

• Advanced AsyncAPI bindings (remaining): Kafka topic-config bindings (partitions, replicas) and v3 operation-level MQTT binding navigation are not yet implemented. MQTT qos/retain, Kafka message key, and AMQP exchange/queue/routing-key bindings are supported (see AsyncAPI Channel Bindings)
• Kafka Protobuf value format: the Confluent wire-format framing (ConfluentWireFormat) is codec-agnostic, but only the Avro codec is wired up. Protobuf (which needs a .proto/descriptor to build DynamicMessage) is deferred. Avro (registry-backed and registry-less) is fully supported for publish and subscribe/verify
• AMQP binding edge cases: operation/message-level AMQP bindings (cc, deliveryMode, etc.) and queue/exchange lifecycle hints (autoDelete, exclusive, vhost) are deferred (see AMQP Channel Bindings). AMQP publish and subscribe/verify are both supported
• Schema Registry authentication (basic auth / API keys): not yet supported — registry-backed mode works against unauthenticated registries only
• Cross-protocol correlation linking (F15): correlating messages across protocols (e.g. HTTP request to Kafka response) is out of scope; each message's correlation ID is self-contained