Capture Message Publish and Drop Event Topic Messages via Data Integration

The Message Publish and Drop Event Topics are system event topics provided by EMQX for real-time monitoring of the internal lifecycle of messages within the Broker, including publication, delivery, acknowledgement, and drop events. These events are crucial for analyzing message flow, diagnosing message loss, and measuring message throughput. Although regular clients cannot directly subscribe to these system topics, you can capture them using the Rule Engine for further processing, such as storing them in databases, forwarding them to other topics, or enabling real-time monitoring and analytics.

Since EMQX Broker functions as an MQTT broker rather than a storage service, and due to security and privacy constraints, it does not store any historical records of message transmission or message drops by default. Therefore, it is recommended to configure Data Integration rules based on message publish and drop event topics. These rules enable you to collect, analyze, and persist message lifecycle information, message reliability metrics, and system delivery quality in real time. This is essential for performance optimization, troubleshooting, and audit-related analysis.

This page explains the use cases of message publish and drop event topics, demonstrates how to capture event messages through the Rule Engine and process them, and shows how to use MQTTX Desktop to simulate clients that trigger these event messages.

Use Cases

Message publish and drop event topics can be applied in a wide range of operational and business scenarios, including:

• Message tracing and delivery path monitoring: Track the full message lifecycle, from publication by the client to successful delivery, helping identify message accumulation, latency, or delivery failures.
• Business metrics and throughput analysis: Count the number of messages published and delivered per client or per topic, providing insights into application load and user behavior patterns.
• Anomaly detection and alerts: Detect message drops caused by missing subscribers, delivery failures, or queue overflows, and capture detailed drop information via event topics.
• Security auditing and access logging: Record which client published which message and whether the message was delivered and consumed normally, supporting audit and compliance requirements.

Trigger Conditions and Key Fields

Message Dropped During Routing ($events/message_dropped)

• Trigger Condition: Triggered when a message has no matching subscribers.
• Key Fields:

Field	Description
clientid	Client ID of the message publisher
username	Username of the message publisher
topic	Message topic
reason	Reason the message was dropped
qos	Message QoS level
timestamp	Timestamp of the drop event (milliseconds)
publish_received_at	Timestamp when the PUBLISH packet reached the Broker (milliseconds)

• Common Drop Reasons:

Reason Code	Description
no_subscribers	No subscribers exist for the topic.
receive_maximum_exceeded	QoS 2 awaiting_rel queue is full.
packet_identifier_inuse	Packet ID (used for QoS 2) is still in use and has not been released.

• Typical Use Case: Monitoring message drops caused by missing subscribers.

Message Dropped During Delivery ($events/delivery_dropped)

• Trigger Condition: Triggered when the subscriber’s message queue is full.
• Key Fields:

Field	Description
from_clientid	Client ID of the message publisher
from_username	Username of the message publisher
topic	Message topic
reason	Reason the message was dropped
qos	Message QoS level
timestamp	Timestamp of the drop event (milliseconds)
publish_received_at	Timestamp when the PUBLISH packet reached the Broker (milliseconds)

• Common Drop Reasons:

Reason Code	Description
no_local	The client is not allowed to receive its own published messages (MQTT 5 No Local = 1).
expired	Message or session expired.
queue_full	Subscriber queue is full (QoS > 0), typically due to slow consumption or long offline periods.
qos0_msg	QoS 0 messages dropped because the queue is full.

• Typical Use Cases:
  • Monitoring subscriber consumption capacity and offline duration
  • Analyzing message backlog or delivery failure causes
  • Using together with $events/message_dropped to fully track message drop behavior across uplink and downlink flows

Message Delivered Event ($events/message_delivered)

• Trigger Condition: Triggered when EMQX successfully sends the message to a subscriber (indicating Broker-side delivery, not client acknowledgement).
• Key Fields:

Field	Description
clientid	Target client ID
from_clientid	Message publisher Client ID
username	Target client's username
topic	Message topic
qos	Message QoS level
timestamp	Timestamp of the event (milliseconds)
publish_received_at	Timestamp when the PUBLISH packet reached the Broker (milliseconds)

• Typical Use Cases:
  • Measuring message delivery success rate
  • Monitoring message delivery latency
  • Combining with $events/message_acked to analyze acknowledgement latency (for QoS 1 and QoS 2)

Message Acknowledged Event ($events/message_acked)

• Trigger Condition: Triggered when the client sends PUBACK or PUBCOMP to acknowledge receipt (QoS 1 and QoS 2 only).
• Key Fields:

Field	Description
id	MQTT message ID, useful for latency tracking
clientid	Target client ID
from_clientid	Publisher client ID
username	Target client's username
topic	Message topic
qos	Message QoS level
timestamp	Timestamp of the event (milliseconds)
publish_received_at	Timestamp when the PUBLISH packet reached the Broker (milliseconds)

• Typical Use Cases:
  1. Computing latency between message delivery and acknowledgement
  2. Analyzing acknowledgement rates for QoS 1 / QoS 2 messages
  3. Ensuring reliable message delivery for mission-critical workflows

TIP

For the full list of fields for event topics, see: MQTT Events.

Configure Data Integration to Capture Drop Event Topic Messages

In real-world deployments, event topics are commonly processed in two ways:

1. Message Republishing: Republishing event messages to another MQTT topic. This method is lightweight, real-time, and compatible with native MQTT workflows.
2. Forwarding to External Services: Sending event messages to external systems such as databases, message queues, or HTTP services. This enables persistent analysis and integration with downstream systems.

This section demonstrates only the message republishing method. For how to forward messages to external services, refer to Forward Message Delivered Event Messages to an HTTP Service.

For other methods of forwarding data to different databases or external services, see the official documentation: EMQX Cloud Data Integration.

Create Message Republish Rule and Action

This section demonstrates how to republish message-dropped event topic messages to another topic.

Create Rule and Action

1. In Data Integration, under Data Forward, select Republish. If connectors already exist, click Create Connector and select Republish.

2. Define the rule SQL in the SQL Editor. For example, to diagnose message dropped events, you can use the following SQL:

   SELECT
       clientid,
       reason,
       topic,
       qos,
       timestamp
   FROM
       "$events/message_dropped"

3. Click Next to add the action.

4. Configure the action:

   • Connector: Use the default Republish.

   • Topic: Set the target topic to message_dropped.

   • Payload, QoS, Retain: Keep default values.

5. Click Confirm to complete the setup.

Test the Rule and Action

You may use MQTTX or any MQTT client.

1. Create two MQTTX connections with ClientIDs pub and sub.
2. sub client subscribes to message_dropped.
3. pub publishes a message to the topic test. The sub client should receive a message such as:

{
    "topic": "test",
    "timestamp": 1761309999897,
    "reason": "no_subscribers",
    "qos": 1,
    "clientid": "pub"
}

This payload indicates that the message was dropped because no subscriber existed for the topic at the time it reached EMQX.

Forward Message Delivered Event Messages to an HTTP Service

This section demonstrates how to forward message delivered event messages to an HTTP service. Before starting, create a VPC Peering Connection to access the connector via private IP, or enable a NAT Gateway to access the connector via public IP.

Create an HTTP Server Connector

1. In Data Integration, under Web Services, choose HTTP Server. If connectors already exist, select Create Connector, then choose HTTP Server.

2. Enter the URL of the target HTTP endpoint and adjust other settings as needed.

   The URL should point to the HTTP service that will receive the forwarded event messages. The connector will send POST requests to this URL using the payload defined in the rule.

3. Click Test to verify connectivity.

4. Click New to complete the connector creation.

Create Rule and Action

1. Click Create Rule.

2. Define the SQL in the SQL Editor, for example:

   SELECT
       from_clientid,
       clientid,
       username,
       payload,
       topic,
       qos
   FROM
       "$events/message_delivered"

3. Click Next to create the action.

4. Choose the previously created HTTP Server connector; keep other settings default.

5. Click Confirm.

TIP

For the full process and configuration of forwarding data to an HTTP service, refer to the official documentation: Ingest MQTT Data into HTTP Server.

Test the Rule and Action

1. Create two MQTTX connections with ClientIDs pub and sub.
2. sub subscribes to the topic test.
3. pub publishes a message to test.
4. The HTTP service should receive a POST request with a payload like:

{
    "username": "test",
    "topic": "test",
    "qos": 0,
    "payload": "Hello Event",
    "from_clientid": "pub",
    "clientid": "sub"
}

Best Practice Recommendations

• For high-frequency events (e.g., message_delivered), enable rate limiting or use a WHEN clause in the rule SQL to filter events, or forward to an external system such as Kafka for buffering.
• Use separate rules and connectors for each event topic or business module to improve maintainability and analysis isolation.
• Combine with client online/offline event topics to achieve full lifecycle monitoring of device connectivity and message behavior.