Connectors
Audience: operators and developers wiring EventMesh to external systems — message queues, databases, chat platforms, AI services. Covers the connector model, the 23 shipped plugins, how to configure and run them. For the API a plugin implements see Connector API split plan; for runtime configuration of the connector scheduler see Configuration reference.
The connector model
A connector moves CloudEvents between EventMesh and an external system on one direction:
- a Source pulls (or receives) data from the external system and the connector runtime publishes it into EventMesh over HTTP;
- a Sink long-polls EventMesh for deliveries and writes them into the external system.
The contract lives in eventmesh-connector-api:
| Method | Source | Sink |
|---|---|---|
init(Properties) | configure from -D flags | configure from -D flags |
resume(String) / poll() | resume from an offset marker; pull the next batch | — |
put(List<CloudEvent>) | — | write a batch (throw on failure → no ACK → redelivery) |
commit(...) | checkpoint after EventMesh accepted the publish | checkpoint after the write |
Delivery semantics are at-least-once: sources checkpoint only after the runtime accepted the publish, and a failing sink write leaves the delivery un-ACKed so EventMesh redelivers (dedup by event id downstream).
Running a connector
Each plugin ships inside the connector-runtime image; bin/start-connector.sh
starts a worker:
# CONNECTOR_OPTS carries the plugin class and every -D flag the plugin reads
CONNECTOR_OPTS="-Dconnector.class=org.apache.eventmesh.connector.kafka.source.KafkaSourceConnector \
-Dconnector.mode=source \
-Dconnector.topic=my-topic \
-DbootstrapServers=broker:9092" \
bin/start-connector.sh
Common flags: eventmesh.runtime.url (default http://localhost:10105),
connector.offset.mode (remote | rocksdb | inmemory), and for multiple
connectors per process the numbered form -Dconnector.1.class=...,
-Dconnector.2.class=... (any -Dconnector.N.* key is passed through to the
plugin's init). The runtime can also schedule connectors dynamically via
/admin/connectors — see the Admin API.
Plugin catalog
Message queues
Web & serverless
Framework bridges
Databases
Storage
Observability
AI & chat
Chat & IM
- DingTalk connector
- Lark / Feishu connector
- Slack connector
- WeChat Official Account connector
- WeCom (WeChat Work) connector
Plugin matrix
| Plugin | Source | Sink | Client dependency |
|---|---|---|---|
| Kafka | ✓ | ✓ | kafka-clients 3.9.0 |
| RocketMQ 4.x | ✓ | ✓ | rocketmq-client |
| RabbitMQ | ✓ | ✓ | amqp-client 5.22.0 |
| Pulsar | ✓ | ✓ | pulsar-client |
| Pravega | ✓ | ✓ | pravega-client 0.11.0 |
| HTTP | ✓ | ✓ | — |
| Knative | ✓ | ✓ | — |
| OpenFunction | ✓ | ✓ | — |
| Spring | ✓ | ✓ | — |
| JDBC | ✓ | ✓ | — |
| Canal (MySQL CDC) | ✓ | ✓ | canal.client 1.1.7 |
| MongoDB | ✓ | ✓ | mongodb-driver-sync 4.11.0 |
| Amazon S3 | ✓ | ✓ | aws-sdk-s3 |
| File | ✓ | ✓ | — |
| Redis | ✓ | ✓ | redisson |
| Prometheus | ✓ | ✓ | — |
| ChatGPT (OpenAI) | ✓ | ✓ | — |
| MCP (Model Context Protocol) | ✓ | ✓ | — |
| DingTalk | ✓ | ✓ | — |
| Lark / Feishu | ✓ | ✓ | — |
| Slack | ✓ | ✓ | — |
| WeChat Official Account | ✓ | ✓ | — |
| WeCom (WeChat Work) | ✓ | ✓ | — |
Adding a plugin
Implement SourceConnector / SinkConnector from eventmesh-connector-api
in a new eventmesh-connector-plugin/eventmesh-connector-<name> module,
register the module in settings.gradle, and model the implementation on an
existing plugin of the same style (pull-based like kafka/jdbc, push-receiver
like http/dingtalk, or writer like rabbitmq/mongodb).