Kafka Bridge

TIP

After EMQX version 3.1, a powerful rule engine is introduced to replace plug-ins. It is recommended that you use it. See Bridge data to Kafka to setup kafka bridges in rule engine.

EMQX bridges and forwards MQTT messages to Kafka cluster:

image

Config file for Kafka bridge plugin: etc/plugins/emqx_bridge_kafka.conf

Configure Kafka Cluster

  1. ## Kafka Server
  2. ## bridge.kafka.servers = 127.0.0.1:9092,127.0.0.2:9092,127.0.0.3:9092
  3. bridge.kafka.servers = 127.0.0.1:9092
  5. ## Kafka Partition Strategy. option value: per_partition | per_broker
  6. bridge.kafka.connection_strategy = per_partition
  8. bridge.kafka.min_metadata_refresh_interval = 5S
  10. ## Produce writes type. option value: sync | async
  11. bridge.kafka.produce = sync
  13. bridge.kafka.produce.sync_timeout = 3S
  15. ## Base directory for replayq to store messages on disk.
  16. ## If this config entry if missing or set to undefined,
  17. ## replayq works in a mem-only manner.
  18. ## i.e. messages are not queued on disk -- in such case,
  19. ## the send or send_sync API callers are responsible for
  20. ## possible message loss in case of application,
  21. ## network or kafka disturbances. For instance,
  22. ## in the wolff:send API caller may trap_exit then
  23. ## react on parition-producer worker pid's 'EXIT'
  24. ## message to issue a retry after restarting the producer.
  25. ## bridge.kafka.replayq_dir = /tmp/emqx_bridge_kafka/
  27. ## default=10MB, replayq segment size.
  28. ## bridge.kafka.producer.replayq_seg_bytes = 10MB
  30. ## producer required_acks. option value all_isr | leader_only | none.
  31. bridge.kafka.producer.required_acks = none
  33. ## default=10000. Timeout leader wait for replicas before reply to producer.
  34. ## bridge.kafka.producer.ack_timeout = 10S
  36. ## default number of message sets sent on wire before block waiting for acks
  37. ## bridge.kafka.producer.max_batch_bytes = 1024KB
  39. ## by default, send max 1 MB of data in one batch (message set)
  40. ## bridge.kafka.producer.min_batch_bytes = 0
  42. ## Number of batches to be sent ahead without receiving ack for the last request.
  43. ## Must be 0 if messages must be delivered in strict order.
  44. ## bridge.kafka.producer.max_send_ahead = 0
  46. ## by default, no compression
  47. # bridge.kafka.producer.compression = no_compression
  49. # bridge.kafka.encode_payload_type = base64
  51. # bridge.kafka.sock.buffer = 32KB
  52. # bridge.kafka.sock.recbuf = 32KB
  53. bridge.kafka.sock.sndbuf = 1MB
  54. # bridge.kafka.sock.read_packets = 20

Configure Kafka Bridge Hooks

  1. ## Bridge Kafka Hooks
  2. ## ${topic}: the kafka topics to which the messages will be published.
  3. ## ${filter}: the mqtt topic (may contain wildcard) on which the action will be performed .
  5. bridge.kafka.hook.client.connected.1     = {"topic": "client_connected"}
  6. bridge.kafka.hook.client.disconnected.1  = {"topic": "client_disconnected"}
  7. bridge.kafka.hook.session.subscribed.1   = {"filter": "#",  "topic": "session_subscribed"}
  8. bridge.kafka.hook.session.unsubscribed.1 = {"filter": "#",  "topic": "session_unsubscribed"}
  9. bridge.kafka.hook.message.publish.1      = {"filter": "#",  "topic": "message_publish"}
  10. bridge.kafka.hook.message.delivered.1    = {"filter": "#",  "topic": "message_delivered"}
  11. bridge.kafka.hook.message.acked.1        = {"filter": "#",  "topic": "message_acked"}

Description of Kafka Bridge Hooks

EventDescription
bridge.kafka.hook.client.connected.1Client connected
bridge.kafka.hook.client.disconnected.1Client disconnected
bridge.kafka.hook.session.subscribed.1Topics subscribed
bridge.kafka.hook.session.unsubscribed.1Topics unsubscribed
bridge.kafka.hook.message.publish.1Messages published
bridge.kafka.hook.message.delivered.1Messages delivered
bridge.kafka.hook.message.acked.1Messages acknowledged

Forward Client Connected / Disconnected Events to Kafka

Client goes online, EMQX forwards ‘client_connected’ event message to Kafka:

  1. topic = "client_connected",
  2. value = {
  3.          "client_id": ${clientid},
  4.          "node": ${node},
  5.          "ts": ${ts}
  6.         }

Client goes offline, EMQX forwards ‘client_disconnected’ event message to Kafka:

  1. topic = "client_disconnected",
  2. value = {
  3.         "client_id": ${clientid},
  4.         "reason": ${reason},
  5.         "node": ${node},
  6.         "ts": ${ts}
  7.         }

Forward Subscription Event to Kafka

  1. topic = session_subscribed
  3. value = {
  4.          "client_id": ${clientid},
  5.          "topic": ${topic},
  6.          "qos": ${qos},
  7.          "node": ${node},
  8.          "ts": ${timestamp}
  9.         }

Forward Unsubscription Event to Kafka

  1. topic = session_unsubscribed
  3. value = {
  4.          "client_id": ${clientid},
  5.          "topic": ${topic},
  6.          "qos": ${qos},
  7.          "node": ${node},
  8.          "ts": ${timestamp}
  9.         }

Forward MQTT Messages to Kafka

  1. topic = message_publish
  3. value = {
  4.          "client_id": ${clientid},
  5.          "username": ${username},
  6.          "topic": ${topic},
  7.          "payload": ${payload},
  8.          "qos": ${qos},
  9.          "node": ${node},
  10.          "ts": ${timestamp}
  11.         }

Forwarding MQTT Message Deliver Event to Kafka

  1. topic = message_delivered
  3. value = {"client_id": ${clientid},
  4.          "username": ${username},
  5.          "from": ${fromClientId},
  6.          "topic": ${topic},
  7.          "payload": ${payload},
  8.          "qos": ${qos},
  9.          "node": ${node},
  10.          "ts": ${timestamp}
  11.         }

Forwarding MQTT Message Ack Event to Kafka

  1. topic = message_acked
  3. value = {
  4.          "client_id": ${clientid},
  5.          "username": ${username},
  6.          "from": ${fromClientId},
  7.          "topic": ${topic},
  8.          "payload": ${payload},
  9.          "qos": ${qos},
  10.          "node": ${node},
  11.          "ts": ${timestamp}
  12.         }

Examples of Kafka Message Consumption

Kafka consumes MQTT clients connected / disconnected event messages:

  1. sh kafka-console-consumer.sh --zookeeper localhost:2181 --topic client_connected --from-beginning
  3. sh kafka-console-consumer.sh --zookeeper localhost:2181 --topic client_disconnected --from-beginning

Kafka consumes MQTT subscription messages:

  1. sh kafka-console-consumer.sh --zookeeper localhost:2181 --topic session_subscribed --from-beginning
  3. sh kafka-console-consumer.sh --zookeeper localhost:2181 --topic session_unsubscribed --from-beginning

Kafka consumes MQTT published messages:

  1. sh kafka-console-consumer.sh --zookeeper localhost:2181 --topic message_publish --from-beginning

Kafka consumes MQTT message Deliver and Ack event messages:

  1. sh kafka-console-consumer.sh --zookeeper localhost:2181 --topic message_delivered --from-beginning
  3. sh kafka-console-consumer.sh --zookeeper localhost:2181 --topic message_acked --from-beginning

TIP

the payload is base64 encoded

Enable Kafka Bridge

  1. ./bin/emqx_ctl plugins load emqx_bridge_kafka