diff --git a/README.md b/README.md index d3b220800a..b403394e4b 100644 --- a/README.md +++ b/README.md @@ -46,7 +46,7 @@ Apache EventMesh is packed with features that help users build event-driven appl **Extensibility & ecosystem** -- **Agent-to-Agent (A2A) collaboration** — a built-in [A2A protocol](docs/a2a-protocol/README_EN.md) turns EventMesh into an agent collaboration bus, bridging synchronous MCP / JSON-RPC 2.0 tool calls and asynchronous event-driven pub/sub for LLM and multi-agent systems. +- **Agent-to-Agent (A2A) collaboration** — a built-in [A2A protocol](docs/eventmesh-a2a-protocol.md) turns EventMesh into an agent collaboration bus, bridging synchronous MCP / JSON-RPC 2.0 tool calls and asynchronous event-driven pub/sub for LLM and multi-agent systems. - **Pluggable storage layer** — [Apache RocketMQ](https://rocketmq.apache.org), [Apache Kafka](https://kafka.apache.org), [Apache Pulsar](https://pulsar.apache.org), [RabbitMQ](https://rabbitmq.com), [Redis](https://redis.io), and more. - **Pluggable interconnector layer** — [connectors](https://github.com/apache/eventmesh/tree/develop/eventmesh-connector-plugin) run as standalone processes acting as the source or sink of SaaS, CloudService, Database, etc. - **Pluggable meta service** — [Consul](https://consulproject.org/en/), [Nacos](https://nacos.io), [ETCD](https://etcd.io), and [Zookeeper](https://zookeeper.apache.org/). @@ -54,6 +54,40 @@ Apache EventMesh is packed with features that help users build event-driven appl - **Powerful event orchestration** through the [Serverless workflow](https://serverlessworkflow.io/) engine. - **Powerful event filtering and transformation.** +## Capability status + +Each EventMesh surface carries an explicit maturity status. The table below is the +**single source of truth** — module-level docs link here instead of restating their +status. See [docs](docs/) for the per-capability guides. + +| Capability | Status | Recommendation | Migration target | +| --- | :---: | --- | --- | +| [HTTP + CloudEvents](docs/eventmesh-client-guide.md) | **GA target** | Recommended — the primary user path (`CloudEventsClient` + `/events/*`) | Primary path | +| [Kafka / RocketMQ storage](eventmesh-storage-plugin/) (4.x, 5.x) | **GA target** | Recommended — pluggable WAL backends, TCK-covered (`MeshStoragePluginTCK`) | Primary path | +| SSE / WebSocket push | **Beta** | Usable — integration-tested; unified ACK/redelivery semantics still landing | Unified push transports | +| Connector Runtime | **Beta** | Usable — 24 connector plugins on the standalone runtime | New connector SPI migration | +| [A2A / Agent Gateway](docs/eventmesh-a2a-protocol.md) | **Experimental** | Evaluate — task store + runtime bridge landed (#5302/#5304); reaper & Meta-backed agent cards pending | Unified Runtime A2A | +| TCP / gRPC / OpenMessaging SDKs | **Legacy-compatible** | Existing users only — kept so old clients run unmodified; not extended | [HTTP + CloudEvents](docs/eventmesh-client-guide.md) | + +Status meanings: + +- **GA target** — feature-complete for the current architecture, integration-tested against real brokers; safe for production. +- **Beta** — functional and tested, but semantics or deployment shape may still shift in a minor release. +- **Experimental** — under active development; APIs and storage layouts may break; wire it up on dev clusters first. +- **Legacy-compatible** — maintained for zero-change compatibility with existing clients; receives fixes but no new features. New integrations should not start here. + +> Migrating off TCP / gRPC SDKs? The legacy clients keep working against the current +> runtime; see the [client guide](docs/eventmesh-client-guide.md) for the +> HTTP + CloudEvents replacement (`CloudEventsClient`). + +### Documentation + +- [Getting started](docs/eventmesh-getting-started.md) — zero to running runtime in minutes +- [Configuration reference](docs/eventmesh-configuration.md) — every runtime key, per-backend settings, security & quota +- [Client guide](docs/eventmesh-client-guide.md) — complete `CloudEventsClient` walkthrough (pub/sub, request-reply, streaming, lite topics) +- [Architecture](docs/eventmesh-architecture.md) — control / data / agent planes; storage SPI; security gate; A2A stack +- [Features](docs/eventmesh-features.md) — feature-by-feature guide (pub/sub, A2A, connectors, security, reliability) + ## Subprojects - [EventMesh-site](https://github.com/apache/eventmesh-site): Apache official website resources for EventMesh. @@ -167,7 +201,7 @@ Upgrade: websocket Connection: Upgrade ``` -> Long-polling, SSE, and WebSocket are interchangeable delivery transports — a subscriber chooses one. SSE and WebSocket are pushed by the server (no polling loop), while long-polling is client-driven. Request-reply (`POST /events/request` + `POST /events/reply`) is also supported. The `CloudEventsClient` Java SDK wraps all of these (`subscribe` / `subscribeSse` / `subscribeWs`); see the [CloudEvents client guide](docs/eventmesh-cloudevents-client-guide.md). +> Long-polling, SSE, and WebSocket are interchangeable delivery transports — a subscriber chooses one. SSE and WebSocket are pushed by the server (no polling loop), while long-polling is client-driven. Request-reply (`POST /events/request` + `POST /events/reply`) is also supported. The `CloudEventsClient` Java SDK wraps all of these (`subscribe` / `subscribeSse` / `subscribeWs`); see the [CloudEvents client guide](docs/eventmesh-client-guide.md). #### 6. Unsubscribing diff --git a/README.zh-CN.md b/README.zh-CN.md index 78810ae9b2..418ce2fa13 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -47,7 +47,7 @@ Apache EventMesh 提供了丰富的能力，帮助用户轻松构建事件驱动 **扩展性与生态** -- **智能体协作（A2A）** —— 内置的 [A2A 协议](docs/a2a-protocol/README.md) 将 EventMesh 打造成智能体协作总线，打通同步的 MCP / JSON-RPC 2.0 工具调用与异步的事件驱动发布订阅，原生支撑大模型（LLM）与多智能体（Multi-Agent）场景。 +- **智能体协作（A2A）** —— 内置的 [A2A 协议](docs/eventmesh-a2a-protocol.md) 将 EventMesh 打造成智能体协作总线，打通同步的 MCP / JSON-RPC 2.0 工具调用与异步的事件驱动发布订阅，原生支撑大模型（LLM）与多智能体（Multi-Agent）场景。 - **可插拔存储层** —— [Apache RocketMQ](https://rocketmq.apache.org)、[Apache Kafka](https://kafka.apache.org)、[Apache Pulsar](https://pulsar.apache.org)、[RabbitMQ](https://rabbitmq.com)、[Redis](https://redis.io) 等。 - **可插拔互联层（Connector）** —— [connectors](https://github.com/apache/eventmesh/tree/develop/eventmesh-connector-plugin) 作为独立进程运行，可充当 SaaS、CloudService、数据库等的 source 或 sink。 - **可插拔元数据服务** —— [Consul](https://consulproject.org/en/)、[Nacos](https://nacos.io)、[ETCD](https://etcd.io) 和 [Zookeeper](https://zookeeper.apache.org/)。 @@ -55,6 +55,38 @@ Apache EventMesh 提供了丰富的能力，帮助用户轻松构建事件驱动 - **强大的事件编排** —— 基于 [Serverless workflow](https://serverlessworkflow.io/) 引擎。 - **强大的事件过滤与转换能力。** +## 能力状态（Capability Status） + +每个 EventMesh 对外能力都有明确的状态定级。下表是**唯一的状态事实来源**——各模块文档链接到这里， +不再各自重复描述。各能力的详细指南见 [docs](docs/)。 + +| 能力 | 状态 | 建议 | 迁移目标 | +| --- | :---: | --- | --- | +| [HTTP + CloudEvents](docs/eventmesh-client-guide.md) | **GA 目标** | 推荐——主用户路径（`CloudEventsClient` + `/events/*`） | 主路径 | +| [Kafka / RocketMQ 存储](eventmesh-storage-plugin/)（4.x、5.x） | **GA 目标** | 推荐——可插拔 WAL 后端，TCK 覆盖（`MeshStoragePluginTCK`） | 主路径 | +| SSE / WebSocket 推送 | **Beta** | 可用——已有集成测试；统一 ACK/重投递语义仍在收敛 | 统一推送传输 | +| Connector Runtime | **Beta** | 可用——独立运行时上的 24 个连接器插件 | 新连接器 SPI 迁移 | +| [A2A / Agent 网关](docs/eventmesh-a2a-protocol.md) | **实验性** | 评估——TaskStore + Runtime 桥已落地（#5302/#5304）；reaper 与 Meta 化 AgentCard 待做 | 统一 Runtime A2A | +| TCP / gRPC / OpenMessaging SDK | **Legacy 兼容** | 仅存量用户——保持老客户端零改动运行；不再扩展 | [HTTP + CloudEvents](docs/eventmesh-client-guide.md) | + +状态含义： + +- **GA 目标** —— 当前架构下功能完整，已对真实 broker 做集成测试；可安全用于生产。 +- **Beta** —— 功能可用且有测试，但语义或部署形态在次版本仍可能调整。 +- **实验性** —— 迭代开发中；API 与存储布局可能破坏性变更；请先在开发集群试用。 +- **Legacy 兼容** —— 仅为存量客户端零改动兼容而维护；只修缺陷、不加功能。新接入不要选这里。 + +> 正在从 TCP / gRPC SDK 迁移？Legacy 客户端在当前 Runtime 上继续可用；替代方案（HTTP + +> CloudEvents 的 `CloudEventsClient`）见[客户端指引](docs/eventmesh-client-guide.md)。 + +### 文档导航 + +- [快速上手（英文）](docs/eventmesh-getting-started.md) —— 零到运行 +- [配置参考（英文）](docs/eventmesh-configuration.md) —— 每个运行时键、后端设置、安全 & 配额 +- [客户端指引（中文）](docs/eventmesh-client-guide.md) —— `CloudEventsClient` 完整用法 +- [架构（英文）](docs/eventmesh-architecture.md) —— 控制面 / 数据面 / 智能体面；存储 SPI；安全闸门；A2A 协议栈 +- [特性（英文）](docs/eventmesh-features.md) —— 逐特性说明（发布订阅、A2A、连接器、安全、可靠性） + ## 子项目 - [EventMesh-site](https://github.com/apache/eventmesh-site): Apache EventMesh 的官方网站资源。 @@ -267,7 +299,7 @@ Upgrade: websocket Connection: Upgrade ``` -> 长轮询、SSE、WebSocket 是可互换的下发传输——订阅者任选其一。SSE 与 WebSocket 由服务端推送（无需轮询循环），长轮询则由客户端驱动。此外还支持请求-应答（`POST /events/request` + `POST /events/reply`）。`CloudEventsClient` Java SDK 封装了以上全部方式（`subscribe` / `subscribeSse` / `subscribeWs`），用法见[CloudEvents 客户端使用指引](docs/eventmesh-cloudevents-client-guide.md)。 +> 长轮询、SSE、WebSocket 是可互换的下发传输——订阅者任选其一。SSE 与 WebSocket 由服务端推送（无需轮询循环），长轮询则由客户端驱动。此外还支持请求-应答（`POST /events/request` + `POST /events/reply`）。`CloudEventsClient` Java SDK 封装了以上全部方式（`subscribe` / `subscribeSse` / `subscribeWs`），用法见[CloudEvents 客户端使用指引](docs/eventmesh-client-guide.md)。 #### 4. 取消订阅 diff --git a/docs/a2a-protocol/README.md b/docs/eventmesh-a2a-protocol.md similarity index 99% rename from docs/a2a-protocol/README.md rename to docs/eventmesh-a2a-protocol.md index d33121fd73..1a68e6e04c 100644 --- a/docs/a2a-protocol/README.md +++ b/docs/eventmesh-a2a-protocol.md @@ -18,6 +18,8 @@ > > Until all three land, treat the gateway as Experimental — wire it up against the > `MetaBackedTaskStore` only on dev clusters. +> +> Canonical maturity level: **Experimental** — see the [capability status table](../README.md#capability-status). --- diff --git a/docs/eventmesh-architecture.md b/docs/eventmesh-architecture.md new file mode 100644 index 0000000000..77bb2fa5f9 --- /dev/null +++ b/docs/eventmesh-architecture.md @@ -0,0 +1,417 @@ +# Apache EventMesh Architecture + +> **Status:** Living document. Reflects the **post-#5296 architecture review** state of the +> `develop` branch (16-issue refactor: #5297–#5306, #5309, #5288). Code locations are +> relative to the module root. Cross-references in the [Documentation](#documentation) table +> link into module-level guides; the **capability status table** in the project README is the +> single source of truth for GA / Beta / Experimental / Legacy tags. + +Apache EventMesh is a **stateless application-layer event mesh** built around three +separation-of-concerns planes: + +1. **Control plane** — subscription / session / state management, cluster coordination, + routing decisions. Stored in a *pluggable meta store* (Consul / Nacos / ETCD / ZK) + and *pluggable storage SPI* (RocksDB / MQ-as-WAL / external KV). +2. **Data plane** — the wire path from a client SDK to a backend MQ, then to subscriber + delivery transports (HTTP long-polling, SSE, WebSocket, request-reply). The + EventMesh Runtime is the only place where delivery semantics live. +3. **Agent plane** — the A2A (Agent-to-Agent) gateway that maps synchronous + tool-call / JSON-RPC semantics onto the same MQ-backed pub/sub substrate, so + multi-agent workloads get durability, observability, and quota from the same + pipeline as ordinary events. + +This page gives the **structural** view. For end-to-end data flow diagrams see +`docs/eventmesh-uni-architecture-redesign.md`. For the security / quota gate, see +the "Security gate" section below and the `#5304` reference docs; for the storage +capability contract, see the `eventmesh-storage-api` TCK. + +--- + +## 1. Top-level decomposition + +``` +┌──────────────────────────────────────────────────────────────────────┐ +│ EventMesh Repository │ +├──────────────────────────────────────────────────────────────────────┤ +│ eventmesh-runtime // the only deployable that ingests │ +│ // and dispatches events │ +│ eventmesh-storage-plugin/ // pluggable MeshStoragePlugin SPI │ +│ eventmesh-storage-api // interface + TCK (StorageCapabilities, │ +│ │ // LiteTopicCapable, MeshStoragePlugin) │ +│ eventmesh-storage-* // backends: RocketMQ, Kafka, Pulsar, │ +│ // RabbitMQ, Redis, … │ +│ eventmesh-connector-plugin/ // 24+ source/sink connector plugins │ +│ eventmesh-connector-runtime // standalone process running them │ +│ eventmesh-protocol-plugin/ // eventmesh-protocol-api + cloudevents│ +│ // + meshmessage + grpc + a2a │ +│ eventmesh-spi // ExtensionFactory, EventMeshSPI, │ +│ // EventMeshExtensionType │ +│ eventmesh-common // shared util / config / metrics │ +│ eventmesh-sdks // client SDKs (Java, …) │ +│ eventmesh-examples // runnable end-to-end examples │ +│ eventmesh-architecture-guard // ArchUnit rules + CI workflow │ +│ eventmesh-agent // reference A2A client / agent core │ +└──────────────────────────────────────────────────────────────────────┘ +``` + +* **One deployable process for the data plane**: `eventmesh-runtime`. It is + *stateless with respect to delivery* (cluster state lives in the meta store and + storage SPI), which means horizontal scale-out is a config change. +* **Storage is a contract, not a driver**: every backend — including the new + RocksDB-local path — implements `MeshStoragePlugin` and passes + `MeshStoragePluginTCK`. The runtime only knows the SPI; it does not import any + vendor JAR. +* **Connectors run in a separate process**: `eventmesh-connector-runtime` (admin + HTTP server + per-connector thread pool). They connect to the runtime over + HTTP+CloudEvents, not in-process, so a misbehaving connector cannot poison the + data plane. +* **Architecture is enforced, not aspirational**: `eventmesh-architecture-guard` + is an ArchUnit test that fails CI if any module violates the layered rules + (e.g. `eventmesh-common` may not depend on `eventmesh-runtime`). See + `#5305` for the FAIL-mode workflow. + +--- + +## 2. Runtime data plane — the wire path + +A single publish, end to end: + +``` +client SDK EventMesh Runtime storage +───────── ───────────────── ─────── +CloudEventsClient.publish(ce) ┌─ UniHttpServer (Netty HTTP/S) OffsetStore + │ POST /events/publish ───────────▶ │ • TlsContextFactory (local KV) + │ │ • RequestContext (security gate) ─┘ + │ │ • SecurityGate.check(ctx, frame) ───┐ + │ │ ├─ FilterChain (auth / acl) │ ACL: AclFilter, + │ │ │ (TokenAuth, Signature, │ TokenAuthFilter, + │ │ │ AclFilter) │ SignatureVerifierFilter + │ │ ├─ QuotaManager (per-tenant) │ Quota: + │ │ └─ AuditSink │ TenantQuotaManager + │ ▼ │ + │ UniIngressService.publish │ + │ │ │ + │ │ EventMeshFrame (single protocol) │ + │ ▼ │ + │ FilterChain.invoke(frame) │ + │ ▼ │ + │ Producer.send(StorageResourceService) ─▶│ ──▶ MQ-as-WAL + │ │ │ + │ │ StorageClient.append(...) │ + │ ▼ │ + │ SendCallback │ + │ ▼ │ + ◀──── HTTP 200 / 4xx (rejected by gate) │ AuditSink.emit(...) ──▶ metrics │ + └─────────────────────────────────────┘ +``` + +A subscribe path mirrors this in reverse. A subscriber's `POST /events/subscribe` +registers a `Subscription` in `ClusterSubscriptionStore` (L2). The runtime's +`LocalDeliverer` polls the storage offset, applies the subscription filter, then +pushes the CloudEvent over the chosen transport (long-poll / SSE / WebSocket / +request-reply). + +The legacy TCP / gRPC / OpenMessaging SDKs still work because the new +`EventMeshFrame` adaptor (`eventmesh-runtime/.../protocol/meshmessage`) and the +`UniTcpServer` (public/internal split per `#5297`) preserve the wire format and +semantics of the old `MeshMessage` / `OpenMessage` clients. + +### Key classes + +| Component | File | Role | +| --- | --- | --- | +| HTTP entry | `eventmesh-runtime/.../http/UniHttpServer.java` | Netty HTTP/S; entry of all `/events/*` + `/a2a/*` traffic; `withSecurityGate(...)` wiring point | +| WebSocket entry | `eventmesh-runtime/.../http/UniWsServer.java` | WebSocket transport for subscribers | +| Ingress orchestrator | `eventmesh-runtime/.../ingress/UniIngressService.java` | Frame-typed facade; single protocol path (`EventMeshFrame`) used by both HTTP and the legacy TCP adaptor | +| Security gate | `eventmesh-runtime/.../security/gate/SecurityGate.java` | Opt-in unified gate (see §4) | +| Filter chain | `eventmesh-runtime/.../security/FilterChain.java` | Auth + ACL filters executed before the gate | +| A2A HTTP handler | `eventmesh-runtime/.../a2a/A2AGatewayHttpHandler.java` | Netty handler for `/a2a/*`; `withSecurityGate(...)` wiring point | +| Connector scheduler | `eventmesh-runtime/.../connector/ConnectorScheduler.java` | Validates `ConnectorDef` against the gate; `withSecurityGate(...)` wiring point | +| Storage SPI | `eventmesh-storage-plugin/eventmesh-storage-api/.../storage/MeshStoragePlugin.java` | Capability-aware contract; backends: `eventmesh-storage-rocketmq`, `eventmesh-storage-kafka`, `eventmesh-storage-pulsar`, `eventmesh-storage-rabbitmq`, `eventmesh-storage-redis`, `eventmesh-storage-rocksdb` | +| Offset store | `eventmesh-runtime/.../offset/` | Local offset tracking; survives restart via RocksDB or via the meta store | +| Delivery state | `eventmesh-runtime/.../state/` | `DeliveryStateStore` (RocksDB) for at-least-once, dead-letter handling | + +--- + +## 3. Control plane — state, sessions, and coordination + +The runtime keeps no delivery state in memory. Every piece of cluster-visible +state lives behind one of the storage SPI interfaces, each with a *clear level* +classification: + +| Level | Purpose | Examples | Stores | +| --- | --- | --- | --- | +| **L1 — local-only** | per-instance delivery hints, cheap to recompute | `OffsetStore` (RocksDB or local KV) | Local file / memory | +| **L2 — cluster-shared** | subscriptions, sessions, agent cards | `SubscriptionStore`, `SessionStore`, `AgentCardRegistry` | Meta store (Nacos / Consul / ETCD / ZK) | +| **L3 — durable-egress** | dead-letter, task records | `DeadLetterStore`, `TaskStore` | Meta store with CAS + fencing epoch | + +The contract for each interface is the same: + +* a `MeshStoragePlugin` implementation in `eventmesh-storage-plugin/*`, +* a thin runtime-side adapter (e.g. `ClusterSubscriptionStore`, + `SessionRegistry`) that adds **fencing tokens / epochs** on top of the meta + store, so split-brain does not corrupt state, +* a TCK (`MeshStoragePluginTCK`) that every backend must pass. + +State changes propagate through a **prefix-watch** pattern on the meta store +(see `eventmesh-runtime/.../cluster/DynamicConfigWatcher.java`). A +`SubscriptionStore.subscribe(...)` call writes to the meta store; every runtime +instance receives the change and re-evaluates its local delivery set. That keeps +the cluster eventually consistent without inventing a Raft implementation. + +For partition-owned delivery, the runtime implements the +`PARTITION_OWNED_PULL` topology (`#5309`): each topic partition is owned by +exactly one runtime, the owner drives the offset, and ownership transfers +are epoch-protected. The owner drives the offset because the storage SPI +itself does not know what is consumed. + +--- + +## 4. Security gate (issue #5304) + +Until 2026 the runtime had no consistent multi-tenant gate. ACL was per-filter +and easy to miss, rate limits were topic-only, and there was no audit trail. +Issue **#5304** added an **opt-in unified gate** that you can wire to the three +ingress points (HTTP, A2A HTTP, Connector admin). + +``` + ingress request + │ + ▼ + SecurityGate.check(RequestContext, EventMeshFrame) + │ + ├─ 1. FilterChain.invoke(frame) ── TokenAuthFilter → SignatureVerifierFilter → AclFilter + │ produces verdict (ALLOW / DENY) + │ + ├─ 2. if ALLOW → QuotaManager.acquire(ctx, Resource) + │ resources: CONNECTIONS | SUBSCRIPTIONS | THROUGHPUT | BACKLOG + │ per-tenant counter; default = UnlimitedQuotaManager (no enforcement) + │ + └─ 3. AuditSink.emit(decision, ctx, frame) ── LoggingAuditSink (default) or custom sink + metrics already cover the aggregation view +``` + +Key design points: + +* **Opt-in, not opt-out.** A `null` gate is a legacy behaviour. New deployments + are expected to install a `TenantQuotaManager` and a custom `AuditSink`. +* **One `RequestContext` per request.** It is immutable, builder-built, and + carries `tenantId`, `principal`, `roles`, `scopes`, `credential`, + `remoteAddress`, `source`, `traceContext`, `quotaKey`, and an `Operation` + enum (`PUBLISH` / `SUBSCRIBE` / `ACK` / `CONNECTOR` / `A2A` / `ADMIN`). +* **`FilterChain` is reused** for the actual auth/ACL verdict, so the gate does + not duplicate policy; it composes it. New filters drop in without touching + the gate. +* **Quota is keyed by `RequestContext.quotaKey()`** (defaults to `tenantId`), + and uses the `Resource` enum so a backend can add custom resources via + extension. + +### Wiring + +* `UniHttpServer#withSecurityGate(SecurityGate)` — gates every `/events/*` + endpoint, including the SSE and WebSocket upgrades. +* `A2AGatewayHttpHandler#withSecurityGate(SecurityGate)` — gates every A2A + endpoint; the A2A operation is recorded in the context so the quota can + distinguish a `tasks/send` from a `tasks/get`. +* `ConnectorScheduler#withSecurityGate(SecurityGate)` — gates + `createConnector(...)` and raises `ConnectorAccessDeniedException` (which + the admin server maps to HTTP 403). + +The **10 unit tests** in `SecurityGateTest` cover each Resource, both +allow/deny verdicts, every Operation, and a smoke test that the gate +short-circuits before the downstream SPI is touched. + +--- + +## 5. Agent plane — A2A on the same substrate + +The A2A (Agent-to-Agent) protocol is an HTTP/JSON contract for invoking a +remote agent as if it were a local function, with a **task** lifecycle +(`submitted → working → completed | failed | canceled`) that maps cleanly onto +a durable `TaskStore`. EventMesh implements the gateway side on top of the +data plane: + +``` + agent A (or client) EventMesh Runtime + ──────────────── ───────────────── + POST /a2a/tasks/send ───────────────▶ A2AGatewayHttpHandler + │ + │ SecurityGate.check(ctx, frame) (Operation.A2A) + ▼ + A2AGatewayService.submit(task) + │ + │ TaskStore.create(task, taskEpoch) ─▶ L3 (meta, CAS) + │ + │ EventMeshA2ATransport.send(target, task) ─▶ topic + │ topic = A2ATopicFactory.agentInbox(target) + ▼ + async fan-out to target agent's listener + │ + ▼ + agent B (or its runtime) processes task + │ + │ POST /a2a/tasks/{id}/reply ─▶ TaskStore.complete(...) + ▼ + completion event published on + A2ATopicFactory.gatewayResponseTopic(...) + │ + ▼ + ◀─── Agent A subscribes (SSE on /a2a/tasks/{id}/stream) ◀── emit +``` + +Key classes: + +* `A2ATopicFactory` — `agentInbox(agentId)`, `gatewayResponseTopic(ns, gw, taskId)`, + and a `+` wildcard for in-process tests. The convention is the same as + Pulsar / MQTT — **single-level** `+`, **not** globstar `*`. +* `A2AGatewayService` — durable task store facade. Every mutation is + epoch-protected: a stale write with a `taskEpoch` different from the create + value is rejected. This is the only safe way to handle out-of-order + responses from multiple agents. +* `AgentCardRegistry` — declares what an agent can do (`/a2a/agent-card`). + Two implementations: `InMemoryAgentCardRegistry` (default) and a + Meta-backed adapter (D2 of the A2A rollout, see issue `#5302`). +* `EventMeshA2ATransport` — bridges the A2A `MessageTransport` SPI to the + same `UniIngressService` the data plane uses, so reliability, quota, and + observability are uniform. + +The protocol is **Experimental** (see capability status table). The wire +contract is the JSON envelope from +[docs/eventmesh-a2a-protocol.md](eventmesh-a2a-protocol.md); the runtime +implementation is documented in +[docs/eventmesh-uni-architecture-redesign.md](eventmesh-uni-architecture-redesign.md). + +--- + +## 6. Connector plane — source / sink as a separate process + +A connector is a long-running process that copies events between EventMesh and +an external system (Kafka topic, Slack channel, MySQL table, S3 bucket, HTTP +endpoint, MCP tool, ChatGPT prompt, …). The `eventmesh-connector-plugin` repo +contains 24 plugin implementations; `eventmesh-connector-runtime` is the +deployable that hosts them. + +``` + external system EventMesh + ────────────── ───────── + source side: ▲ + ┌──────────┐ poll / listen ┌──────────────────┐ │ + │ Kafka │ ───────────────▶ │ Sink side: │ │ + │ MySQL │ │ SinkConnector │ │ + │ S3 │ │ (CloudEvent in) │ │ + │ file │ └────────┬─────────┘ │ + └──────────┘ │ │ + ▼ │ + EventMesh HTTP │ + /events/publish │ + ───┘ + ▲ + │ + EventMesh HTTP │ + /events/subscribe ───┘ + │ + ▼ + Source side: + SourceConnector + │ + ▼ + external write + (HTTP / JDBC / gRPC / …) +``` + +* `ConnectorDef` is the immutable, validated description of one connector: + `id`, `className`, `mode` (`source` / `sink` / `source-sink`), + `topic`, `clientId`, `sinkClass`. New fields are added via the + `ConnectorDef.Builder` — never by setter injection. +* `ConnectorScheduler` owns the lifecycle (`init → start → stop`) and + registers each connector against the SecurityGate. A misconfigured + connector is rejected **before** the JAR is loaded, so a plugin bug + cannot take down the runtime. +* Connectors talk to the runtime over **HTTP + CloudEvents**, never + in-process. They can be deployed, scaled, and restarted independently. + +See `eventmesh-connector-runtime/.../ConnectorManager.java` for the +lifecycle and `eventmesh-connector-runtime/.../ConnectorAdminServer.java` +for the admin HTTP surface. + +--- + +## 7. Module boundary rules (architecture-guard) + +`eventmesh-architecture-guard` is an ArchUnit test suite that **fails the +build** if a module violates the layering rules. It runs in two places +(issue `#5305`): + +* locally on `./gradlew :eventmesh-architecture-guard:check` so a developer + gets a 30-second feedback loop, and +* on every PR via `.github/workflows/architecture-guard.yml` (Ubuntu-only) + with a path filter to skip irrelevant pushes. + +The current rule set covers: + +* `eventmesh-common` may not depend on `eventmesh-runtime` (and vice versa + for any runtime-only class). +* `eventmesh-storage-api` may not depend on any concrete backend. +* `eventmesh-protocol-plugin/*` is partitioned into `public` (SPI) and + `internal` (implementation) — the `public → internal` direction is + allowed; the reverse is not. +* `eventmesh-connector-plugin/*` plugins may not depend on + `eventmesh-runtime`; they only see the CloudEvents HTTP contract. +* `eventmesh-agent` is allowed to depend on `eventmesh-sdks` and + `eventmesh-protocol-api` only. + +If a future change needs a cross-module dependency, the rule must be +**amended in code review**, not bypassed. The 30-second local loop and the +PR workflow are both in place to make this discipline cheap. + +--- + +## 8. Cross-cutting concerns + +### Metrics and observability + +* Prometheus metrics under `org.apache.eventmesh.metrics.*` cover the + gate (allowed/denied counts per Operation, per Resource), the runtime + (publish/subscribe rate, queue depth), and the connectors (per-connector + throughput, error rate). +* Trace context (`RequestContext.traceContext`) propagates through every + plane so a single `traceId` follows an event from the client SDK to + the storage backend and back out to the subscriber. +* The audit sink is the place to add custom per-tenant audit + destinations (Kafka, S3, …). + +### Configuration + +* Runtime configuration is documented in + [docs/eventmesh-configuration.md](eventmesh-configuration.md). New + keys are added behind the existing prefixes (`eventmesh.runtime.*`, + `eventmesh.security.gate.*`, `eventmesh.storage.*`, + `eventmesh.connector.*`) — never under a global `eventmesh.*` root. +* Per-backend settings are isolated (e.g. + `eventmesh.storage.rocketmq.namesrvAddr` is **not** visible to a + Kafka backend). The runtime boots a backend only if its class is on + the classpath. + +### Build & test + +* Gradle multi-module; `:eventmesh-runtime:assemble` produces the + runnable distribution. +* `./gradlew :eventmesh-architecture-guard:check` runs the ArchUnit + rules before unit tests. +* Integration tests use Testcontainers (Nacos, RocketMQ) — not + embedded servers — to catch backend-version drift. + +--- + +## Documentation + +| Page | What it covers | +| --- | --- | +| [docs/eventmesh-getting-started.md](eventmesh-getting-started.md) | Zero-to-running runtime; per-backend quickstarts | +| [docs/eventmesh-configuration.md](eventmesh-configuration.md) | Every runtime key, security & quota, per-backend overrides | +| [docs/eventmesh-client-guide.md](eventmesh-client-guide.md) | `CloudEventsClient` walkthrough: pub/sub, request-reply, SSE, WebSocket, lite topics | +| [docs/eventmesh-a2a-protocol.md](eventmesh-a2a-protocol.md) | A2A wire contract and task lifecycle | +| [docs/production-readiness.md](production-readiness.md) | Verified capabilities, SLOs, runbooks | +| [docs/eventmesh-uni-architecture-redesign.md](eventmesh-uni-architecture-redesign.md) | End-to-end flow diagrams and the redesign rationale | +| [docs/eventmesh-offset-lb-frame-design.md](eventmesh-offset-lb-frame-design.md) | `EventMeshFrame` design (single protocol path, `#5299`) | +| [docs/eventmesh-features.md](eventmesh-features.md) | Feature-by-feature guide (the companion to this page) | diff --git a/docs/eventmesh-client-guide.md b/docs/eventmesh-client-guide.md new file mode 100644 index 0000000000..eebd029570 --- /dev/null +++ b/docs/eventmesh-client-guide.md @@ -0,0 +1,685 @@ +# Apache EventMesh Client Guide + +> **Audience:** application developers using the `eventmesh-sdk-java` `CloudEventsClient` +> and `A2AClient` to publish / subscribe / stream / dispatch A2A tasks. +> +> **Backend-agnostic.** This guide covers the full API surface and is the single +> source of truth for what the client SDK does. The capability status table in +> the project README is the source of truth for GA / Beta / Experimental / Legacy +> tags; backend-specific configuration is in +> [docs/eventmesh-configuration.md](eventmesh-configuration.md); architectural +> context is in [docs/eventmesh-architecture.md](eventmesh-architecture.md). + +The new architecture ships with **one client SDK** that exposes two surface APIs: + +* `CloudEventsClient` — HTTP + [CloudEvents 1.0](https://cloudevents.io) pub/sub, + request-reply, SSE / WebSocket push, RocketMQ 5.x Lite Topic, LLM streaming. + This is the **primary user path** and the recommended way to integrate with + EventMesh. +* `A2AClient` — Agent-to-Agent task dispatch on top of the same Runtime + ([docs/eventmesh-a2a-protocol.md](eventmesh-a2a-protocol.md)). Used by multi-agent systems that + need a durable task lifecycle. + +Both clients talk **only to the EventMesh Runtime** (HTTP). The underlying +storage backend — RocketMQ 4.x, RocketMQ 5.x, Kafka, or any other +`MeshStoragePlugin` implementation — is **completely transparent** to the client. +Switching backends is a server-side configuration change, not a client change. + +> Legacy `EventMeshHttpClient` / `EventMeshTCPClient` are kept for protocol +> compatibility but are no longer extended. New integrations must use +> `CloudEventsClient` (or `A2AClient` for A2A workloads). See the +> [migration notes](#15-legacy-compatibility) at the end of this guide. + +--- + +## Table of contents + +1. [Quick orientation](#1-quick-orientation) +2. [`CloudEventsClient` API reference](#2-cloudeventsclient-api-reference) +3. [Builder and configuration](#3-builder-and-configuration) +4. [Publish / subscribe patterns](#4-publish--subscribe-patterns) +5. [Request-reply (synchronous RPC)](#5-request-reply-synchronous-rpc) +6. [Subscriber transports: long-poll / SSE / WebSocket](#6-subscriber-transports-long-poll--sse--websocket) +7. [RocketMQ 5.x Lite Topic](#7-rocketmq-5x-lite-topic) +8. [LLM streaming call (Mode 1 / Mode 2)](#8-llm-streaming-call-mode-1--mode-2) +9. [Security: tokens, signatures, the unified gate](#9-security-tokens-signatures-the-unified-gate) +10. [Reliability: ACK, retries, dead-letter, idempotency](#10-reliability-ack-retries-dead-letter-idempotency) +11. [`A2AClient` for agent workloads](#11-a2aclient-for-agent-workloads) +12. [Backend selection: RocketMQ 4.x / 5.x / Kafka](#12-backend-selection-rocketmq-4x--5x--kafka) +13. [End-to-end example](#13-end-to-end-example) +14. [Operational checklist](#14-operational-checklist) +15. [Legacy compatibility](#15-legacy-compatibility) +16. [Code locations](#16-code-locations) + +--- + +## 1. Quick orientation + +A typical client looks like this: + +```java +CloudEventsClient client = CloudEventsClient.builder() + .runtimeUrl("http://localhost:8080") // EventMesh Runtime HTTP endpoint + .clientId("order-svc") // unique per JVM + .pollIntervalMs(500L) // long-poll cadence + .build(); + +client.subscribe("orders", "BROADCAST", event -> { + System.out.println("got " + event.getId() + " type=" + event.getType()); +}); + +CloudEvent e = CloudEventsClient.event( + "evt-1", "order-svc", "order.created", "{\"amt\":99}".getBytes(StandardCharsets.UTF_8)); +client.publish("orders", e); +``` + +That's the entire model. The rest of this document covers edge cases and the +optional surfaces (A2A, streaming, security, Lite Topic). + +--- + +## 2. `CloudEventsClient` API reference + +```java +org.apache.eventmesh.client.cloudevents.CloudEventsClient +``` + +| Method | Returns | Notes | +| --- | --- | --- | +| `builder()` | `CloudEventsClientBuilder` | Entry point — see §3 | +| `publish(topic, CloudEvent)` | `boolean` | Single publish; 202 on success | +| `publish(topic, List)` | `boolean` | Batched publish (HTTP body batching) | +| `request(topic, CloudEvent, timeoutMs)` | `CloudEvent` | **Blocking** request-reply; `null` on timeout; late replies are dropped | +| `reply(correlationId, CloudEvent)` | `boolean` | Reply side of request-reply; uses the `emcorrelationid` extension | +| `subscribe(topic, mode, Consumer)` | — | Long-poll subscribe; **handler return ⇒ auto-ACK** | +| `subscribeWithAck(topic, mode, Predicate)` | — | Long-poll subscribe; `Predicate` returns `true` = ACK, `false` = no-ACK (at-least-once re-delivery on dispatcher timeout) | +| `subscribeSse(topic, mode, Consumer)` | — | SSE push subscribe; runs over HTTP on `/events/stream` | +| `subscribeWs(topic, mode, Consumer)` | — | WebSocket push subscribe; needs `wsUrl` (separate port) | +| `unsubscribe(topic)` | — | Unsubscribe one topic; stops the long-poll loop if no topics remain | +| `unsubscribe()` | — | Unsubscribe all; stop all loops / pushes | +| `createLiteTopic(parent, lite)` | `boolean` | **RocketMQ 5.x only.** Idempotent create of a Lite Topic (RIP-83) | +| `publishLite(parent, lite, CloudEvent)` | `boolean` | **5.x only.** Publish to LMQ inside the Lite Topic | +| `subscribeLite(parent, lite, Consumer)` | — | **5.x only.** Background poll loop; offset managed inside the storage plugin (no ACK / no DLQ) | +| `unsubscribeLite(parent, lite)` | — | **5.x only.** Stop the background poll loop for one Lite Topic | +| `streaming()` | `StreamingOperations` | Entry point for LLM streaming — see §8 | +| `shutdown()` | — | Stop everything: long-poll, SSE, WS, Lite loops, streaming sessions | +| `static event(id, source, type, byte[] data)` | `CloudEvent` | Convenience constructor | + +### Subscriber mode (`DistributionMode`) + +`mode` is one of the string constants from +`org.apache.eventmesh.runtime.subscription.DistributionMode`: + +| Mode | Semantics | +| --- | --- | +| `BROADCAST` | Every subscriber gets every message | +| `LOAD_BALANCE` | Each message goes to exactly one subscriber in the group | +| `MULTICAST` | Multi-cast delivery | +| `LOAD_BALANCE_STICKY` | Hash by `partitionKey` extension for stable affinity (preserves per-key order) | + +--- + +## 3. Builder and configuration + +```java +CloudEventsClient.builder() + .runtimeUrl("http://localhost:8080") // required — Runtime HTTP endpoint + .clientId("my-service") // required — unique per JVM + .pollIntervalMs(500L) // long-poll cadence (default: builder default) + .wsUrl("http://localhost:8082") // optional — required for subscribeWs + .build(); +``` + +| Builder key | Required | Default | Notes | +| --- | --- | --- | --- | +| `runtimeUrl` | yes | — | HTTP(S) base URL of the Runtime | +| `clientId` | yes | — | Used in subscription registration, `SubscriptionStore`, quota key | +| `pollIntervalMs` | no | builder default | Long-poll cadence. Larger value = more idle time per round; smaller = more requests | +| `wsUrl` | no (yes for WS) | — | **WebSocket** endpoint. Runtime exposes WS on a separate port (configured at server start) | + +Environment variables are honored via `System.getProperty` for tests: + +```java +CloudEventsClient client = CloudEventsClient.builder() + .runtimeUrl(System.getProperty("eventmesh.runtime.url", "http://localhost:8080")) + .clientId("demo-" + System.currentTimeMillis()) + .build(); +``` + +TLS / mTLS is configured at the **Runtime** (server side), not the client. The +client just talks to `https://...` once TLS is enabled. See +[docs/eventmesh-configuration.md](eventmesh-configuration.md#security). + +--- + +## 4. Publish / subscribe patterns + +### 4.1 Auto-ACK subscribe (the simple case) + +```java +client.subscribe("orders", "BROADCAST", event -> { + System.out.println("got " + event.getId()); + // any thrown exception still counts as ACK — use subscribeWithAck for at-least-once +}); +``` + +### 4.2 Manual ACK for at-least-once + +```java +client.subscribeWithAck("orders", "LOAD_BALANCE", event -> { + try { + process(event); // your business logic + return true; // ACK → offset advances + } catch (Exception ex) { + return false; // no-ACK → re-delivery on dispatcher timeout + } +}); +``` + +Business idempotency is your responsibility. EventMesh guarantees +**at-least-once**, not exactly-once. + +### 4.3 Batch publish + +```java +List batch = ...; +boolean ok = client.publish("orders", batch); +``` + +The Runtime splits the batch into per-partition writes inside the storage plugin. +A single failure inside the batch surfaces as `false` and the Runtime returns +`502` for that call. + +--- + +## 5. Request-reply (synchronous RPC) + +The request-reply pattern uses the **`emcorrelationid`** CloudEvents extension +(all-lowercase, no hyphens — CloudEvents disallows hyphens in extension names). + +```java +// requester +CloudEvent req = CloudEventsClient.event("req-1", "caller", "query.price", payload); +CloudEvent reply = client.request("price-req", req, 10_000L); // up to 10s +if (reply != null) { /* use reply */ } + +// replier +responder.subscribe("price-req", "LOAD_BALANCE", event -> { + Object corr = event.getExtension("emcorrelationid"); + if (corr != null) { + CloudEvent r = CloudEventsClient.event("reply-1", "price-svc", "query.price.reply", + priceJson(event).getBytes(StandardCharsets.UTF_8)); + responder.reply(corr.toString(), r); + } +}); +``` + +`request(...)` is **blocking** on the client thread. Late replies arriving after +`timeoutMs` are dropped at the Runtime. Use `subscribeSse` or `subscribeWs` +when you need to keep the channel open. + +--- + +## 6. Subscriber transports: long-poll / SSE / WebSocket + +All three transports produce the same `CloudEvent` payload to the handler; the +only difference is the **push direction**. + +| Transport | Endpoint | Push direction | Port | +| --- | --- | --- | --- | +| Long-poll | `POST /events/subscribe` | client-driven | Runtime HTTP port (default 8080) | +| SSE | `GET /events/stream` (text/event-stream) | server push | Runtime HTTP port (default 8080) | +| WebSocket | runtime WS endpoint | server push, bi-directional | Runtime WS port (default 8082, configurable) | + +WebSocket needs a separate port because the WS upgrade is a different protocol +negotiation than plain HTTP. The Runtime starts the WS server on its own port +(server-side configuration), and the client **must** configure `wsUrl` +explicitly. Pointing `wsUrl` at the HTTP port will fail the WS handshake. + +```java +// SSE — same port as HTTP +client.subscribeSse("orders", "BROADCAST", event -> { /* server-push */ }); + +// WebSocket — separate port +CloudEventsClient wsClient = CloudEventsClient.builder() + .runtimeUrl("http://localhost:8080") // HTTP (publish / long-poll / SSE) + .wsUrl("http://localhost:8082") // WS push + .clientId("ws-sub").build(); +wsClient.subscribeWs("orders", "BROADCAST", event -> { /* WS push */ }); +``` + +All three transports auto-ACK on handler return (like `subscribe`). Use the +manual-ACK variant only for the long-poll transport. + +--- + +## 7. RocketMQ 5.x Lite Topic + +Lite Topic (RIP-83) is RocketMQ 5.5+'s hierarchical message container. A Lite +Topic lives inside a normal parent topic; the parent must be declared `LITE` +type, then individual `lite` queues inside it share the parent's storage +budget. Useful for **session / sub-class** fan-out at very high cardinality. + +> **Backend-only feature.** `createLiteTopic` / `publishLite` / `subscribeLite` +> return `false` (or no callbacks fire) on RocketMQ 4.x, Kafka, or any +> non-`LiteTopicCapable` storage backend. The Runtime returns `501 Not +> Implemented` for the corresponding endpoints. + +```java +// 1. Declare — idempotent, call once at startup +client.createLiteTopic("orders", "user-42"); + +// 2. Subscribe — background poll loop, push-style callback (no ACK, no DLQ) +client.subscribeLite("orders", "user-42", event -> { /* process lite event */ }); + +// 3. Publish — routes to LMQ via __LITE_TOPIC property +client.publishLite("orders", "user-42", + CloudEventsClient.event("lt-1", "order-svc", "order.lite", payload)); +``` + +Differences from ordinary topic subscribe: + +* **No ACK / no DLQ** — offset is managed inside the storage plugin. At-least-once + is best-effort. +* **Client-driven polling** — `subscribeLite` runs a `GET /events/lite/poll` loop + on the client side. Use `unsubscribeLite(parent, lite)` to stop one Lite + subscription; `unsubscribe()` / `shutdown()` stop everything. + +--- + +## 8. LLM streaming call (Mode 1 / Mode 2) + +EventMesh provides two streaming patterns for LLM-style use cases (token +chunks flowing back, multi-turn conversation context). + +| Mode | Use case | Direction | Entry | +| --- | --- | --- | --- | +| **Mode 1** — streaming call | client → agent (LLM), agent streams tokens back | request/response, push | `client.streaming().openSession(...)` | +| **Mode 2** — pub/sub on a session | producer writes chunks; consumer reads via SSE | publish/subscribe | `client.subscribeSession(sessionId)` / `client.openSessionPublisher(sessionId)` | + +### 8.1 Mode 1 — single call + +```java +CloudEventsClient client = CloudEventsClient.builder() + .runtimeUrl("http://localhost:8080").clientId("my-app").build(); + +try (StreamingResponse r = client.streaming() + .openSession(OpenSession.builder().clientId(client.clientId()).build()) + .call("Introduce EventMesh in three sentences")) { + r.forEach(chunk -> System.out.print(chunk.getChunk())).join(); +} +``` + +`forEach` fires once per token (or delta). `.join()` blocks until the stream +ends. Closing the `StreamingResponse` ends **one round**; it does **not** close +the session. + +### 8.2 Mode 1 — multi-turn conversation + +```java +StreamingSession session = client.streaming() + .openSession(OpenSession.builder().clientId("my-app").build()); +try { + try (StreamingResponse r1 = session.call("I'm Zhang San, a Java engineer")) { + r1.forEach(c -> System.out.print(c.getChunk())).join(); + } + // session remembers the previous round + try (StreamingResponse r2 = session.call("What's my name and job?")) { + r2.forEach(c -> System.out.print(c.getChunk())).join(); + } +} finally { + session.close(); +} +``` + +Multi-turn context is owned by the agent's `ConversationStore`, keyed by +`sessionId`. + +### 8.3 Mode 2 — pub/sub on a session + +Useful when chunks need to be **persistent** (durable across process restarts) +or **fan-out** to multiple consumers. Internally uses the storage plugin's +Lite Topic. + +```java +// consumer side +StreamingResponse sub = client.subscribeSession("my-session-id"); +sub.forEach(chunk -> System.out.println("[" + chunk.getSeq() + "] " + chunk.getChunk())).join(); +sub.close(); + +// producer side +SessionPublisher pub = client.openSessionPublisher("my-session-id"); +pub.publish("Hello", false); // non-terminal frame +pub.publish(" world", false); +pub.publish("", true); // terminal frame — consumer's forEach completes +pub.close(); +``` + +### 8.4 Implementing an agent + +An agent that participates in Mode 1 follows a four-step contract: + +1. parse `sessionId`, `prompt`, `replyTo` from the inbound CloudEvent +2. on each LLM token → emit a non-terminal frame `{chunk: token, done: false}` +3. on normal completion → emit a terminal frame `{chunk: "", done: true}` +4. on error → emit a terminal error frame `{chunk: "", done: true, error: "..."}` + +Reference implementation: `eventmesh-agent/.../StreamingAgent.java` +(instantiate with an LLM client, an `agentParent` topic, the agent's `agentId`, +and a `ConversationStore`). + +### 8.5 Server-side configuration for streaming + +The Runtime pre-creates the agent / client parent topics. For Mode 2 also +pre-create `sessionStreamParent`. The 6-arg `SessionRouter` enables +`sessionTtlMs` + `sessionStreamParent`; the 4-arg variant is Mode 1 only. + +--- + +## 9. Security: tokens, signatures, the unified gate + +By default, the Runtime is open. Production deployments **must** enable the +unified security gate (issue #5304) on the server. From the client side, the +only practical change is that you may need to attach credentials as HTTP +headers / CloudEvents extensions: + +* `Authorization: Bearer ` — picked up by the built-in `TokenAuthFilter` + and recorded into the `RequestContext` as `principal` / `scopes` +* CloudEvents extension `emtenantid` — drives per-tenant quota in + `TenantQuotaManager` +* CloudEvents extension `emcorrelationid` — request-reply correlation + +The gate runs `FilterChain` (TokenAuth → SignatureVerifier → Acl) → +`QuotaManager` (per-`Resource` counter, default `UnlimitedQuotaManager`) → +`AuditSink` (default `LoggingAuditSink`) on every ingress. The +`Operation` enum is recorded in the context, so quota can distinguish a +`publish` from a `subscribe` from an A2A call. + +For configuration and the three wiring points +(`UniHttpServer.withSecurityGate`, `A2AGatewayHttpHandler.withSecurityGate`, +`ConnectorScheduler.withSecurityGate`) see +[docs/eventmesh-configuration.md](eventmesh-configuration.md#security) and +[docs/eventmesh-architecture.md §4](eventmesh-architecture.md#4-security-gate-issue-5304). + +> **The client SDK does not need to "know" about the gate.** A deployment that +> enables the gate is a server-side change. The client sends the same CloudEvent +> and the Runtime decides. If the Runtime requires auth, it returns `401` and +> your handler can re-authenticate and retry. + +--- + +## 10. Reliability: ACK, retries, dead-letter, idempotency + +| Concept | Where it lives | Client responsibility | +| --- | --- | --- | +| At-least-once | Runtime `DeliveryStateStore` | Use `subscribeWithAck` and return `true` only after success | +| Retries | Runtime retry policy | `false` from your predicate triggers a re-delivery after the dispatcher timeout | +| Dead-letter | Runtime `DeadLetterStore` | Inspect / replay via admin endpoints (see [eventmesh-configuration.md](eventmesh-configuration.md#admin)) | +| Idempotency | — | **You.** Use `event.getId()` as the dedup key. The Runtime does not deduplicate. | +| Offset | Runtime `OffsetStore` (L1) | — | +| Subscription state | Runtime `SubscriptionStore` (L2) | Survives Runtime restart via the meta store | +| Task state (A2A) | Runtime `TaskStore` (L3) | — | + +For the storage-state taxonomy (L1 / L2 / L3) and the +`MeshStoragePlugin` / `MeshStoragePluginTCK` contract, see +[docs/eventmesh-architecture.md §3](eventmesh-architecture.md#3-control-plane--state-sessions-and-coordination). + +Configuration knobs: see +[docs/eventmesh-configuration.md](eventmesh-configuration.md#deliverystate) +(`eventmesh.runtime.delivery.*`). + +--- + +## 11. `A2AClient` for agent workloads + +For multi-agent systems, the [A2A protocol](eventmesh-a2a-protocol.md) gives you a +durable task lifecycle (`submitted → working → completed | failed | +canceled`) on top of the same storage substrate. The client side is +`org.apache.eventmesh.protocol.a2a.A2AClient` (in the +`eventmesh-protocol-a2a` module). + +### 11.1 Builder + +```java +A2AClient client = A2AClient.builder() + .gatewayUrl("http://localhost:8080") // Runtime A2A gateway (port 8080 by default) + .namespace("default") + .agentName("order-agent") + .heartbeatInterval(30_000L) + .build(); +``` + +| Builder key | Required | Notes | +| --- | --- | --- | +| `gatewayUrl` | yes | Runtime HTTP base URL (A2A is served on the same HTTP port) | +| `namespace` | recommended | A2A namespace for topic isolation | +| `agentName` | recommended | Local agent identity; used in topic factory and `AgentCard` | +| `heartbeatInterval` | no | Heartbeat to the Runtime; default 30s | +| `socketTimeoutMs` | no | Underlying HTTP client socket timeout | + +### 11.2 Core operations + +| Method | Returns | Notes | +| --- | --- | --- | +| `sendTask(task)` | `TaskResult` | Submit a task; returns immediately with `taskId` + initial state | +| `sendTaskSync(task, timeoutMs)` | `TaskResult` | Submit and block until terminal state (or timeout) | +| `sendTaskAsync(task, Consumer)` | — | Submit and stream intermediate states via callback | +| `getTaskStatus(taskId)` | `TaskResult` | Re-query the current state of a task | +| `cancelTask(taskId)` | `boolean` | Request cancellation; the target agent stops work if it can | +| `streamTaskStatus(taskId, Consumer)` | — | SSE push of state transitions until terminal | +| `listAgents()` | `List` | Browse the agent registry | +| `registerAgentCard(AgentCard)` | `boolean` | Publish this agent's capability description | + +`TaskResult` exposes `taskId`, `state`, `data`, `error`, and `targetAgent`. + +### 11.3 Idempotency: `taskEpoch` + +Each task has a `taskEpoch` field that is **set at creation and never reset**. +Stale writes with a `taskEpoch` different from the create value are rejected by +the Runtime. Use the same `taskEpoch` across retries so the same logical task +always lands in the same slot. + +### 11.4 Server-side wiring + +The A2A gateway is enabled at the Runtime by booting the `A2AGatewayServer` +(Netty) on a configurable port (defaults to the main HTTP port). The endpoint +surface is: + +* `POST /a2a/tasks/send` — submit a task +* `POST /a2a/tasks/sync` — submit and block +* `GET /a2a/tasks/{id}` — query state +* `POST /a2a/tasks/{id}/cancel` — cancel +* `GET /a2a/tasks/{id}/stream` — SSE stream of state transitions +* `GET /a2a/agents` — list agents +* `POST /a2a/agents` — register an agent card + +See [docs/eventmesh-a2a-protocol.md](eventmesh-a2a-protocol.md) for the wire contract and +[docs/eventmesh-architecture.md §5](eventmesh-architecture.md#5-agent-plane--a2a-on-the-same-substrate) +for the runtime architecture. + +--- + +## 12. Backend selection: RocketMQ 4.x / RocketMQ 5.x / Kafka + +**The client code is identical across backends.** Switching from one storage +backend to another is a **Runtime** configuration change; the same +`CloudEventsClient` (and `A2AClient`) bytes run unchanged. + +### 12.1 Server-side configuration matrix + +| | RocketMQ 4.x | RocketMQ 5.x | Kafka | +| --- | --- | --- | --- | +| Plugin SPI key | `rocketmq` | `rocketmq5` | `kafka` | +| Storage module | `eventmesh-storage-plugin/eventmesh-storage-rocketmq` | `eventmesh-storage-plugin/eventmesh-storage-rocketmq5` | `eventmesh-storage-plugin/eventmesh-storage-kafka` | +| Connection | `NettyRemotingClient` direct (no `rocketmq-client` JAR) | Same — pure 5.5 remoting | `kafka-clients` (assign+seek+poll, **no consumer group**; EventMesh owns offsets) | +| Auth | ACL (optional) | ACL (optional) | SASL/SSL pass-through (`security.protocol` / `sasl.mechanism` / `sasl.jaas.config` are passed verbatim to `kafka-clients`) | +| Lite Topic | — | yes (`LiteTopicCapable`) | — | + +Per-backend keys are listed in +[docs/eventmesh-configuration.md](eventmesh-configuration.md#storage). +Pick one `eventmesh.storage.type` at Runtime startup: + +```bash +# 4.x +EVENTMESH_STORAGE_TYPE=rocketmq EVENTMESH_ROCKETMQ_NAMESRV=127.0.0.1:9876 bin/start.sh + +# 5.x +EVENTMESH_STORAGE_TYPE=rocketmq5 EVENTMESH_ROCKETMQ5_NAMESRV=127.0.0.1:9876 bin/start.sh + +# Kafka +EVENTMESH_STORAGE_TYPE=kafka bin/start.sh # eventmesh.properties has bootstrap + SASL +``` + +### 12.2 Client-visible behavioral differences + +The HTTP contract is the same — but the storage-plugin choices have +client-visible consequences for **subscription semantics** under failure: + +| Dimension | RocketMQ 4.x | RocketMQ 5.x | Kafka | +| --- | --- | --- | --- | +| Consumption model | Classic PULL (EventMesh owns offset + partition ownership) | **POP** (broker allocates queues + lease gate) | assign + seek + poll (no consumer group; EventMesh owns offset) | +| Multi-instance de-dup | EventMesh `PartitionOwnership` | broker POP + lease | EventMesh `PartitionOwnership` (Kafka assign) | +| Offset ACK semantics | offset advances only on ACK | same | same (Kafka offset not committed; EventMesh-managed) | +| `publish` / `subscribe` / `request` / `reply` | consistent | consistent | consistent | +| Lite Topic | not supported | supported | not supported | + +The `publish` / `subscribe` / `subscribeWithAck` / `request` / `reply` API +contract is identical across all three backends — that is the point of the +abstraction. + +### 12.3 Kafka + SASL example + +For SASL-enabled Kafka clusters (e.g. wemq-kafka), set in `eventmesh.properties`: + +```properties +eventMesh.server.kafka.namesrvAddr=127.0.0.1:9094 +security.protocol=SASL_PLAINTEXT +sasl.mechanism=PLAIN +sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; +``` + +The `KafkaMeshStoragePlugin` forwards `security.*` / `sasl.*` / `ssl.*` keys +verbatim to the underlying `KafkaProducer` / `KafkaConsumer` / `AdminClient`. +Plain-text Kafka clusters need none of these. + +--- + +## 13. End-to-end example + +```java +public class Demo { + public static void main(String[] args) throws Exception { + CloudEventsClient client = CloudEventsClient.builder() + .runtimeUrl(System.getProperty("eventmesh.runtime.url", "http://localhost:8080")) + .clientId("demo-" + System.currentTimeMillis()) + .pollIntervalMs(500L) + .build(); + + client.subscribeWithAck("demo-topic", "LOAD_BALANCE", event -> { + System.out.println("processing: " + event.getId() + " type=" + event.getType()); + return true; // ACK + }); + + for (int i = 0; i < 10; i++) { + CloudEvent e = CloudEventsClient.event( + "e" + i, "demo", "demo.tick", + ("tick-" + i).getBytes(StandardCharsets.UTF_8)); + client.publish("demo-topic", e); + } + + Thread.sleep(60_000L); + client.shutdown(); + } +} +``` + +Switching the backend is a server-side change only: + +```bash +# 4.x +EVENTMESH_STORAGE_TYPE=rocketmq EVENTMESH_ROCKETMQ_NAMESRV=127.0.0.1:9876 bin/start.sh + +# 5.x (Lite Topic capable) +EVENTMESH_STORAGE_TYPE=rocketmq5 EVENTMESH_ROCKETMQ5_NAMESRV=127.0.0.1:9876 bin/start.sh + +# Kafka (SASL in eventmesh.properties) +EVENTMESH_STORAGE_TYPE=kafka bin/start.sh +``` + +The same `Demo` class runs unchanged on all three. + +--- + +## 14. Operational checklist + +| Check | Where | What to look for | +| --- | --- | --- | +| Runtime reachable | client log | First `publish` returns `true`; first `subscribe` callback fires within `pollIntervalMs` | +| `clientId` uniqueness | Runtime log | A `clientId` collision prints a warning; use a different `clientId` per JVM | +| Backend connection | Runtime startup log | `[storage] connected to ` line; otherwise no subscriptions will fire | +| Security gate | Runtime response | `401` on first request → auth header missing; `429` → quota exhausted; `403` → ACL denied | +| Quota exhaustion | Runtime metrics | `eventmesh_security_gate_quota_*` per-tenant counters | +| Dead-letter inspection | admin HTTP (port 8081) | `GET /admin/dlq?topic=` | +| A2A agent registry | `A2AClient.listAgents()` | Should return at least one `AgentCard` for `agentName` you registered | + +See [docs/production-readiness.md](production-readiness.md) for SLOs and +runbooks. + +--- + +## 15. Legacy compatibility + +The legacy `EventMeshHttpClient` and `EventMeshTCPClient` continue to work +against the current Runtime, but they are **legacy-compatible** in the +capability status table and are not extended. + +| Old client | New client | Migration | +| --- | --- | --- | +| `EventMeshHttpClient.publish(CloudEventMessage)` | `CloudEventsClient.publish(topic, CloudEvent)` | Switch the event from `CloudEventMessage` to `CloudEvent`; topic is a string | +| `EventMeshTCPClient.subscribe(topic, EventListener)` | `CloudEventsClient.subscribe(topic, mode, Consumer)` | Add a `mode`; switch the callback to `Consumer` | +| TCP subscribe with custom `Session` | WebSocket | WebSocket is the modern bi-directional transport | +| OpenMessaging SDK | CloudEventsClient | The OpenMessaging wire is not supported in the new Runtime; use the HTTP + CloudEvents path | + +For TCP / gRPC SDK migration details, see the legacy-compat section of this guide (§15) — this document is the +authoritative home for the new client API; the old guide is preserved in git +history for the migration notes. + +--- + +## 16. Code locations + +* `CloudEventsClient` — `eventmesh-sdks/eventmesh-sdk-java/.../cloudevents/CloudEventsClient.java` + + `CloudEventsClientBuilder` +* Streaming — `eventmesh-sdks/eventmesh-sdk-java/.../cloudevents/stream/` (operations, + response, session, request, publisher, exception) +* `A2AClient` — `eventmesh-protocol-plugin/eventmesh-protocol-a2a/.../A2AClient.java` +* `A2ATopicFactory` — `eventmesh-protocol-plugin/eventmesh-protocol-a2a/.../A2ATopicFactory.java` + (`agentInbox(agentId)`, `gatewayResponseTopic(ns, gw, taskId)`, `+` wildcard) +* Runtime HTTP entry — `eventmesh-runtime/.../http/UniHttpServer.java` + (`/events/*` endpoints; `withSecurityGate(...)` wiring point) +* A2A HTTP handler — `eventmesh-runtime/.../a2a/A2AGatewayHttpHandler.java` + (`/a2a/*` endpoints; `withSecurityGate(...)` wiring point) +* Streaming agent — `eventmesh-agent/.../StreamingAgent.java` +* Storage plugins — + * `eventmesh-storage-plugin/eventmesh-storage-rocketmq/` (SPI key `rocketmq`) + * `eventmesh-storage-plugin/eventmesh-storage-rocketmq5/` (SPI key `rocketmq5`, + `LiteTopicCapable`) + * `eventmesh-storage-plugin/eventmesh-storage-kafka/` (SPI key `kafka`, + assign+seek+poll, SASL pass-through) +* Security gate — `eventmesh-runtime/.../security/gate/` (`SecurityGate`, + `RequestContext`, `QuotaManager`, `AuditSink`, `GateDecision`) +* Architecture — `eventmesh-architecture-guard/.../guard/ArchitectureRules.java` + (ArchUnit layered-architecture enforcement) + +See also: + +* [docs/eventmesh-architecture.md](eventmesh-architecture.md) — system + architecture, control / data / agent planes +* [docs/eventmesh-features.md](eventmesh-features.md) — feature-by-feature + guide +* [docs/eventmesh-configuration.md](eventmesh-configuration.md) — every + runtime key +* [docs/eventmesh-getting-started.md](eventmesh-getting-started.md) — + zero-to-running guide +* [docs/eventmesh-a2a-protocol.md](eventmesh-a2a-protocol.md) — A2A wire contract +* [docs/production-readiness.md](production-readiness.md) — SLOs, runbooks diff --git a/docs/eventmesh-cloudevents-client-guide.md b/docs/eventmesh-cloudevents-client-guide.md deleted file mode 100644 index cf0bfb4887..0000000000 --- a/docs/eventmesh-cloudevents-client-guide.md +++ /dev/null @@ -1,421 +0,0 @@ -# EventMesh CloudEvents 客户端使用指引（RocketMQ 4.x / 5.x / Kafka 后端通用） - -## 1. 这是什么 - -新架构下 EventMesh 对外**只提供一套客户端 SDK**：`eventmesh-sdk-java` 的 `CloudEventsClient`。 -它是 **HTTP + CloudEvents** 的极简客户端（无 TCP/gRPC、无 MQ 组语义），通过 EventMesh Runtime 的 -`/events/*` HTTP 端点收发标准 [CloudEvents](https://cloudevents.io) 事件。 - -**关键点：客户端与 MQ 后端完全解耦。** 客户端只跟 EventMesh Runtime（HTTP）打交道，Runtime 后面接 -RocketMQ 4.x、RocketMQ 5.x 还是 Kafka，对客户端代码**完全透明**——同一份客户端代码可以原样跑在任一 -后端上。三种后端的区别只在 **Runtime（服务端）的部署配置**，不在客户端。 - -> 旧的 `EventMeshHttpClient` / `EventMeshTCPClient`（legacy）仍保留作老协议兼容；新接入请用 `CloudEventsClient`。 - ---- - -## 2. 客户端 API 总览 - -```java -org.apache.eventmesh.client.cloudevents.CloudEventsClient -``` - -| 方法 | 说明 | -|---|---| -| `builder()` | 构造器入口（见下） | -| `publish(topic, CloudEvent)` → `boolean` | 发布单条（202=成功） | -| `publish(topic, List)` → `boolean` | 批量发布 | -| `request(topic, CloudEvent, timeoutMs)` → `CloudEvent` | **阻塞式请求-应答**（等回复） | -| `reply(correlationId, CloudEvent)` → `boolean` | 应答方回送一个 reply | -| `subscribe(topic, mode, Consumer)` | 长轮询订阅，**handler 返回后自动 ACK** | -| `subscribeWithAck(topic, mode, Predicate)` | 长轮询订阅，**手动 ACK**：predicate 返回 `true`=已处理(ACK，offset 推进)；`false`=不 ACK（dispatcher 超时后重投，at-least-once） | -| `createLiteTopic(parent, lite)` → `boolean` | **（仅 5.x）** 建/声明 lite topic（确保 parent 为 LITE 类型） | -| `publishLite(parent, lite, CloudEvent)` → `boolean` | **（仅 5.x）** 发布到 lite topic（路由进 LMQ） | -| `subscribeLite(parent, lite, Consumer)` | **（仅 5.x）** 订阅 lite topic（后台循环拉 LMQ + handler 回调，push 风格，与 `subscribe` 一致；无 ACK，offset 自管） | -| `subscribeSse(topic, mode, Consumer)` | SSE 推送订阅（`/events/stream`） | -| `subscribeWs(topic, mode, Consumer)` | WebSocket 推送订阅 | -| `unsubscribe(topic)` | 退订单个 topic（服务端移除该 `{clientId, topic}` 订阅，其他保留；若无普通 topic 剩余则停长轮询循环） | -| `unsubscribeLite(parent, lite)` | **（仅 5.x）** 停某个 lite 订阅的后台拉取循环（lite 无服务端注册，纯客户端停循环） | -| `unsubscribe()` | 退订全部（服务端按 clientId 移除所有订阅 + 停所有循环/推送） | -| `shutdown()` | 关闭客户端（停轮询/SSE/WS） | -| `static event(id, source, type, byte[] data)` | 便捷构造一个 CloudEvent | - -**Builder：** -```java -CloudEventsClient.builder() - .runtimeUrl("http://localhost:8080") // EventMesh Runtime 地址（必填） - .clientId("my-service") // 客户端标识（必填） - .pollIntervalMs(500L) // 长轮询间隔（默认由 builder 决定） - .build(); -``` - -**订阅模式 `mode`**（`org.apache.eventmesh.runtime.subscription.DistributionMode`，传字符串）： -- `BROADCAST` — 广播：每个订阅者都收到全量消息。 -- `LOAD_BALANCE` — 负载均衡：同一消息只投递给组内一个订阅者。 -- `MULTICAST` — 多播。 -- `LOAD_BALANCE_STICKY` — 按 partition key 稳定哈希到同一个订阅者（保序）。 - ---- - -## 3. 快速上手 - -### 依赖 -客户端 jar：`eventmesh-sdk-java`（Maven 坐标 `org.apache.eventmesh:eventmesh-sdk-java`，或直接用本仓库 -`eventmesh-sdks/eventmesh-sdk-java` 模块）。**客户端不依赖任何 MQ jar**（不引 RocketMQ / kafka-clients）——它只发 HTTP。 - -### 发布 + 订阅（长轮询，自动 ACK） -```java -CloudEventsClient client = CloudEventsClient.builder() - .runtimeUrl("http://localhost:8080") - .clientId("order-svc") - .pollIntervalMs(500L) - .build(); - -// 订阅（后台长轮询，每条事件 handler 返回后自动 ACK） -client.subscribe("orders", "BROADCAST", event -> { - System.out.println("收到: " + event.getId()); -}); - -// 发布 -CloudEvent e = CloudEventsClient.event("evt-1", "order-svc", "order.created", - "{\"amt\":99}".getBytes(StandardCharsets.UTF_8)); -boolean ok = client.publish("orders", e); -``` - -### 手动 ACK（at-least-once，业务幂等窗口由客户端控制） -```java -client.subscribeWithAck("orders", "LOAD_BALANCE", event -> { - try { - process(event); // 业务处理 - return true; // true → ACK，offset 推进 - } catch (Exception ex) { - return false; // false → 不 ACK，ACK 超时后重投 - } -}); -``` - -### 请求-应答（阻塞） -```java -// 请求方：发一个请求，阻塞等回复（带 emcorrelationid） -CloudEvent req = CloudEventsClient.event("req-1", "caller", "query.price", payload); -CloudEvent reply = client.request("price-req", req, 10_000L); // 最多等 10s -if (reply != null) { /* 用 reply */ } - -// 应答方：订阅请求 topic，看到带 correlation 的请求就 reply -responder.subscribe("price-req", "LOAD_BALANCE", event -> { - Object corr = event.getExtension("emcorrelationid"); - if (corr != null) { - CloudEvent r = CloudEventsClient.event("reply-1", "price-svc", "query.price.reply", - priceJson(event).getBytes(StandardCharsets.UTF_8)); - responder.reply(corr.toString(), r); - } -}); -``` - -> 注意：correlation 用 CloudEvents 扩展名 **`emcorrelationid`**（全小写无连字符，CloudEvents 规范不允许扩展名含连字符）。 - -### Lite Topic（仅 5.x 后端） - -RocketMQ 5.5 Lite Topic（RIP-83）：topic 内的二级消息容器。客户端三步走（仅对 5.x 后端有效；4.x 后端返回 `false`、服务端 501）： - -```java -// 1. 建/声明 lite topic（确保 parent 为 LITE 类型）—— 幂等，首次调用一次即可 -client.createLiteTopic("orders", "user-42"); - -// 2. 订阅 lite topic（后台循环拉 LMQ + handler 回调，push 风格，和 subscribe 一致） -client.subscribeLite("orders", "user-42", event -> { /* 处理 lite 事件 */ }); - -// 3. 发布到 lite topic（带 __LITE_TOPIC，broker 路由进 LMQ） -client.publishLite("orders", "user-42", - CloudEventsClient.event("lt-1", "order-svc", "order.lite", payload)); -``` - -> Lite topic 语义与普通 topic 不同：`(parent, lite)` 唯一标识一个 LMQ 容器；`subscribeLite` 走**后台拉取**（不走 -> EventMesh 的 ACK/重投/DLQ 可靠层，offset 在存储插件内自管）。适合海量轻量会话/子分类场景。停止用 `unsubscribe()` / `shutdown()`。 - -### SSE / WebSocket 推送 -```java -// SSE：复用 HTTP 端口的 /events/stream（text/event-stream） -client.subscribeSse("orders", "BROADCAST", event -> { /* 服务端长连接推送 */ }); - -// WebSocket：runtime 的 WS 服务跑在独立端口（启动参数 -Deventmesh.ws.port=，0=自动）， -// 客户端需显式指定 wsUrl（否则会连到 HTTP 端口的 SSE 端点，握手失败） -CloudEventsClient wsClient = CloudEventsClient.builder() - .runtimeUrl("http://localhost:8080") // HTTP 端口（publish / subscribe / SSE） - .wsUrl("http://localhost:8082") // WS 推送端口 - .clientId("ws-sub").build(); -wsClient.subscribeWs("orders", "BROADCAST", event -> { /* WS 推送 */ }); -``` -> SSE 与 WS 都是服务端推送（客户端不用轮询）；WS 需独立端口 + `wsUrl`，SSE 走 HTTP 端口。两者都自动 ACK。 - -### 关闭 -```java -client.unsubscribe(); -client.shutdown(); -``` - ---- - -## 4. 后端选择：RocketMQ 4.x / RocketMQ 5.x / Kafka - -**结论：客户端代码完全一致。** 三种后端的差异 100% 在 Runtime（服务端）侧——你只是把 Runtime 指向不同的 -MQ 集群，客户端那份 `CloudEventsClient` 代码一行都不用改。 - -### 4.1 唯一要改的：Runtime（服务端）配置 - -| | RocketMQ 4.x | RocketMQ 5.x | Kafka | -|---|---|---|---| -| 选插件 | `-Deventmesh.storage.type=rocketmq` | `-Deventmesh.storage.type=rocketmq5` | `-Deventmesh.storage.type=kafka` | -| 接入地址 | `eventMesh.server.rocketmq.namesrvAddr=<4.x:9876>` | `eventMesh.server.rocketmq5.namesrvAddr=<5.x:9876>` | `eventMesh.server.kafka.namesrvAddr=` | -| 鉴权 | 无 / ACL | 无 / ACL | **SASL**：`security.protocol` / `sasl.mechanism` / `sasl.jaas.config`（透传给 kafka-clients） | -| 插件实现 | `RocketMQRemotingStoragePlugin`（4.9 remoting） | `RocketMQ5RemotingStoragePlugin`（5.5 remoting） | `KafkaMeshStoragePlugin`（kafka-clients） | -| 连接方式 | `NettyRemotingClient` 直连（remoting，**不引 rocketmq-client**） | 同左（**不引 rocketmq-client/gRPC**） | `KafkaProducer`/`Consumer`/`AdminClient`（assign+seek+poll，**无 consumer group**，EventMesh 自管 offset） | - -部署时在 `eventmesh.properties`（或 `-D` 参数）里设这些值，Runtime 启动时按 `eventmesh.storage.type` -SPI 加载对应插件。客户端只要把 `runtimeUrl` 指向该 Runtime 的 HTTP 端口即可。 - -**Kafka + SASL（如 wemq-kafka）配置示例**（`eventmesh.properties`）： -```properties -eventMesh.server.kafka.namesrvAddr=127.0.0.1:9094 -security.protocol=SASL_PLAINTEXT -sasl.mechanism=PLAIN -sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; -``` -> Kafka 的 `security.*` / `sasl.*` / `ssl.*` 配置由 `KafkaMeshStoragePlugin` 透传给 producer / consumer / AdminClient（`kafka.` 前缀也接受）。非 SASL 的明文 Kafka 不设这些即可。 - -### 4.2 对客户端透明的行为差异（了解即可，不影响 API） - -| 维度 | RocketMQ 4.x | RocketMQ 5.x | Kafka | -|---|---|---|---| -| 消费模型 | 经典 PULL（EventMesh 自管 offset + 分区所有权） | **POP**（broker 端分配队列，poll-all） | **assign+seek+poll**（无 consumer group，EventMesh 自管 offset） | -| 多实例去重 | EventMesh `PartitionOwnership` + 分区分配 | broker POP + lease gate | EventMesh `PartitionOwnership`（Kafka assign 分区） | -| ACK 语义 | offset 仅 ACK 后推进 | 相同 | 相同（Kafka offset 不 commit，EventMesh 自管） | -| 普通 publish/subscribe/request-reply | 一致 | 一致 | 一致 | - -客户端调用的 `publish` / `subscribe` / `subscribeWithAck` / `request` / `reply` / `ack`，行为契约在三种后端上 -完全一致——这就是抽象的价值。 - -### 4.3 Lite Topic（仅 5.x） - -RocketMQ 5.5 的 **Lite Topic（RIP-83）** 是 5.x 独有能力（topic 内的二级消息容器，面向海量会话/子分类）。 -**已端到端接入 HTTP 客户端**（仅对 5.x 后端可用；4.x / kafka / standalone 后端调 lite 方法返回 `false`，服务端 501）： - -- 客户端：`CloudEventsClient.createLiteTopic(parent, lite)` / `publishLite(parent, lite, event)` / `subscribeLite(parent, lite, handler)`。 -- Runtime 端点：`POST /events/lite/create`、`POST /events/lite/publish`、`GET /events/lite/poll`（→ `UniIngressService` → `LiteTopicCapable` 存储插件；`subscribeLite` 在客户端后台循环调 `/events/lite/poll`）。 -- 插件：`RocketMQ5RemotingStoragePlugin` 实现 `LiteTopicCapable`（`sendLite` 带 `__LITE_TOPIC` 消息属性路由进 LMQ；`pullLite` 经典 PULL + liteTopic 从 LMQ 拉，offset 自管）。 - -Lite 用法见 §3.5。**注意**：4.x 后端没有 lite 能力，调 `publishLite` / `subscribeLite` 会得到 `false`/无推送（服务端返回 501）。 - ---- - -## 5. 一份端到端示例（4.x / 5.x 通用客户端） - -```java -// === 这份代码同时适用于 4.x 后端和 5.x 后端，无需改动 === -public class Demo { - public static void main(String[] args) { - CloudEventsClient client = CloudEventsClient.builder() - .runtimeUrl(System.getProperty("eventmesh.runtime.url", "http://localhost:8080")) - .clientId("demo-" + System.currentTimeMillis()) - .pollIntervalMs(500L) - .build(); - - client.subscribeWithAck("demo-topic", "LOAD_BALANCE", event -> { - System.out.println("处理: " + event.getId() + " type=" + event.getType()); - return true; // ACK - }); - - for (int i = 0; i < 10; i++) { - CloudEvent e = CloudEventsClient.event("e" + i, "demo", "demo.tick", - ("tick-" + i).getBytes(StandardCharsets.UTF_8)); - client.publish("demo-topic", e); - } - - // Runtime.sleep / 处理 … - client.shutdown(); - } -} -``` - -切后端时**只改服务端**： -```bash -# 4.x 后端 -gradle :eventmesh-runtime:dist -EVENTMESH_STORAGE_TYPE=rocketmq EVENTMESH_ROCKETMQ_NAMESRV=127.0.0.1:9876 bin/start.sh - -# 5.x 后端（同一个 dist 镜像，三个 storage 插件都在） -EVENTMESH_STORAGE_TYPE=rocketmq5 EVENTMESH_ROCKETMQ5_NAMESRV=127.0.0.1:9876 bin/start.sh - -# Kafka 后端（bootstrap + SASL 在 eventmesh.properties 里配，见 §4.1） -EVENTMESH_STORAGE_TYPE=kafka bin/start.sh -``` -（具体环境变量名以 `bin/start.sh` 的 `-D` 映射为准；`eventmesh.storage.type` / `eventMesh.server..namesrvAddr` 是关键属性。） - ---- - -## 6. 注意事项 / 限制 - -- **ACK 语义**：`subscribe`（自动 ACK）在 handler 抛异常时**不会**重投（已 ACK）；要 at-least-once，用 - `subscribeWithAck` 并在失败时返回 `false`。 -- **CloudEvents 扩展名**不能含连字符：`emcorrelationid`、`emsignature`、`emdlqreason`、`emtenantid` 等（全小写）。 -- **请求-应答超时**：`request(topic, event, timeoutMs)` 是阻塞调用，超时返回 `null`；迟到回复会被丢弃。 -- **Lite Topic**：仅 5.x 后端（已端到端接入 HTTP 客户端，见 §4.3）；4.x / Kafka 后端调 lite 方法返回 `false`（服务端 501）。 -- **后端切换**：RocketMQ 4.x / 5.x / Kafka 之间切换是服务端配置切换，客户端代码无需改动、无需重新打包。 -- **Kafka SASL**：对带鉴权的 Kafka（如 wemq-kafka），在 `eventmesh.properties` 配 `security.protocol` / `sasl.mechanism` / `sasl.jaas.config`，插件透传给 kafka-clients（见 §4.1）。 - ---- - -## 附：相关代码位置 -- 客户端：`eventmesh-sdks/eventmesh-sdk-java/.../client/cloudevents/CloudEventsClient.java`（+ `CloudEventsClientBuilder`） -- Runtime HTTP 端点：`eventmesh-runtime/.../http/UniHttpServer.java`（`/events/*`） -- 4.x 存储插件：`eventmesh-storage-plugin/eventmesh-storage-rocketmq`（SPI key `rocketmq`） -- 5.x 存储插件：`eventmesh-storage-plugin/eventmesh-storage-rocketmq5`（SPI key `rocketmq5`，含 `LiteTopicCapable`） -- Kafka 存储插件：`eventmesh-storage-plugin/eventmesh-storage-kafka`（SPI key `kafka`，assign+seek+poll + SASL 透传） -- 设计文档：`docs/eventmesh-uni-architecture-redesign.md`（§20–§21 流式调用设计） - ---- - -# 流式调用使用指南（整合 sdk-streaming-call-guide） - -> 面向**应用开发者**：如何用 EventMesh 客户端 SDK 做 LLM 风格的**流式调用**（token 逐片返回、支持多轮上下文）。 - -## 7. 流式调用：它是什么 - -流式调用让客户端发起一次请求后，**持续接收一串文本分片（token / delta）**，直到收到结束标记——和 OpenAI Chat Completions 的 `stream: true`、SSE 打字机效果一致。 - -EventMesh 提供两种流式模式： - -| 模式 | 用途 | 入口 | 角色 | -|------|------|------|------| -| **Mode 1 — 流式调用** | 客户端发起请求，Agent（接 LLM）逐 token 回复 | `openSession` → `call` → `forEach` | HTTP/SSE 客户端（无 MQ 依赖） | -| **Mode 2 — 发布/订阅** | 生产者往 session 的 MQ 通道写 chunks，消费者通过 SSE 订阅读取 | `subscribeSession` / `openSessionPublisher` | 生产者 + 消费者（解耦、跨进程、跨时间） | - -**前置条件**：一个已运行的 EventMesh Runtime（默认 `http://localhost:8080`），其上注册了至少一个流式 Agent（Mode 1）或启用了 Mode 2 的发布/订阅。 - ---- - -## 8. Mode 1：流式调用（30 秒快速开始） - -```java -import org.apache.eventmesh.client.cloudevents.CloudEventsClient; -import org.apache.eventmesh.client.cloudevents.stream.*; - -CloudEventsClient client = CloudEventsClient.builder() - .runtimeUrl("http://localhost:8080") - .clientId("my-app") - .build(); - -// 开一个 session，发一轮，收 token，关 session -StreamingSession session = client.streaming() - .openSession(OpenSession.builder().clientId(client.clientId()).build()); -try { - try (StreamingResponse r = session.call("用三句话介绍 EventMesh")) { - r.forEach(chunk -> System.out.print(chunk.getChunk())).join(); - } -} finally { - session.close(); -} -``` - -`forEach` 每收到一个分片就回调一次，`.join()` 阻塞到流结束。 - ---- - -## 9. Mode 1：多轮会话（Session） - -```java -StreamingSession session = client.streaming() - .openSession(OpenSession.builder().clientId("my-app").build()); - -try { - System.out.println("sessionId = " + session.sessionId()); - - // 第一轮 - try (StreamingResponse r1 = session.call("我叫张三，是一名 Java 工程师")) { - r1.forEach(chunk -> System.out.print(chunk.getChunk())).join(); - } - - // 第二轮（同一 session，Agent 记得上一轮） - try (StreamingResponse r2 = session.call("我叫什么名字？做什么工作？")) { - r2.forEach(chunk -> System.out.print(chunk.getChunk())).join(); - } -} finally { - session.close(); // 最终销毁会话（幂等） -} -``` - -**三条极易踩的语义**： -1. 关闭某一轮的 `StreamingResponse` 只结束这一轮的读取，**不会关闭 session**。 -2. 多轮上下文由 Agent 端的 `ConversationStore` 按 `sessionId` 维护。 -3. `OpenSession` 有 `clientId`（必填）、`model`（可选）；`StreamRequest` 有 `prompt`（必填）、`model`（可选）、`timeout`（可选）。 - ---- - -## 10. Mode 2：发布/订阅 - -Mode 2 把 session 的流式数据**外化到 MQ**（LiteTopic），实现**跨进程/跨时间**的流式消费。 - -### 10.1 订阅（消费） - -```java -StreamingResponse sub = client.subscribeSession("my-session-id"); -sub.forEach(chunk -> { - System.out.println("[" + chunk.getSeq() + "] " + chunk.getChunk()); -}).join(); -sub.close(); -``` - -### 10.2 发布（生产） - -```java -SessionPublisher pub = client.openSessionPublisher("my-session-id"); -pub.publish("Hello", false); // 非终止帧 -pub.publish(" world", false); -pub.publish("", true); // 终止帧 → 订阅者的 forEach 完成 -pub.close(); -``` - ---- - -## 11. Mode 1：实现一个流式 Agent - -Agent 收到请求后契约四步： -1. 解析 `sessionId`、`prompt`、`replyTo` -2. 每产生一个 token → 发非终止帧 `{chunk: token, done: false}` -3. 流正常结束 → 发终止帧 `{chunk: "", done: true}` -4. 出异常 → 发终止错误帧 `{chunk: "", done: true, error: "..."}` - -参考实现 `eventmesh-agent/.../StreamingAgent.java` 接 OpenAI 兼容 LLM： - -```java -StreamingAgent agent = new StreamingAgent(agentClient, agentParent, agentId, llm, conversations); -agent.start(); // 订阅 agent 控制信道 -control.ready(agentId); // 向 runtime 报告就绪 -``` - ---- - -## 12. 部署配置 - -### Mode 1（流式调用） - -预创建 agent + client parent topics，配 `AgentAnchoredStrategy`，4 参 SessionRouter。 - -### Mode 2（发布/订阅，可选） - -额外预创建 `sessionStreamParent`，用 6 参 SessionRouter（`sessionTtlMs` + `sessionStreamParent`）。 - -两种模式可同时启用（入口不同，互不干扰）。 - ---- - -## 13. 流式调用常见问题 - -**Q: Mode 1 和 Mode 2 怎么选？** -需要 Agent 承载 LLM、多轮对话上下文 → Mode 1。需要跨进程/跨时间消费流式数据、崩溃恢复 → Mode 2。 - -**Q: 报 `IllegalStateException: posture already active`？** -同一个 `StreamingResponse` 只能用一次 `forEach`。 - -**Q: 如何排错"收不到响应"？** -① Runtime 是否起、Agent 是否 `ready` + 心跳；② `/session/open` 是否返回 200 且有 `agentId`；③ parent 是否预创建为 LiteTopic；④ broker 是否开了 LiteTopic。 diff --git a/docs/eventmesh-configuration.md b/docs/eventmesh-configuration.md new file mode 100644 index 0000000000..05c1b15eaf --- /dev/null +++ b/docs/eventmesh-configuration.md @@ -0,0 +1,139 @@ +# EventMesh Configuration Reference + +Every configuration key for the EventMesh Runtime, in one place. Defaults live in +[`eventmesh-runtime/conf/eventmesh.properties`](../eventmesh-runtime/conf/eventmesh.properties); +precedence is **system property (`-D`) > environment variable > properties file**. + +> Status of each capability (GA / Beta / Experimental / Legacy): see the +> [capability status table](../README.md#capability-status). + +--- + +## 1. Storage backend selection + +| Key | Env override | Values | Default | +|---|---|---|---| +| `eventmesh.storage.type` | `EVENTMESH_STORAGE_TYPE` | `rocketmq` / `rocketmq5` / `kafka` | `kafka` | + + + +## 2. Storage backends + +### RocketMQ 4.x (`rocketmq`) + +| Key | Default | Description | +|---|---|---| +| `eventMesh.server.rocketmq.namesrvAddr` | `127.0.0.1:9876` | Name server address list | +| `eventMesh.server.rocketmq.cluster` | `DefaultCluster` | Cluster name for topic routing | + +Direct remoting connections (no `rocketmq-client` dependency on the classpath). + +### RocketMQ 5.x (`rocketmq5`) + +| Key | Default | Description | +|---|---|---| +| `eventMesh.server.rocketmq5.namesrvAddr` | `127.0.0.1:9876` | Name server address list | +| `eventMesh.server.rocketmq5.cluster` | `DefaultCluster` | Cluster name | +| `eventmesh.rocketmq5.lite.checkpoint.interval.ms` | `5000` | Lite-topic pull-offset checkpoint interval; `<= 0` persists only on shutdown. Bounds crash replay to at most one interval of messages | + +Enables 5.x **POP consumption** and **Lite Topic** (ordered per-lite-queue messaging; see the +client guide §lite-topics). + +### Kafka (`kafka`) + +| Key | Default | Description | +|---|---|---| +| `eventMesh.server.kafka.namesrvAddr` | `localhost:9092` | Bootstrap servers | +| `security.protocol` | — | `SASL_PLAINTEXT` etc. for secured clusters | +| `sasl.mechanism` | — | `PLAIN`, `SCRAM-SHA-512`, ... | +| `sasl.jaas.config` | — | Full JAAS login module line | +| `ssl.*` | — | Any kafka-clients SSL key (transparently forwarded) | + +EventMesh uses `assign` + `seek` + `poll` — **no consumer groups**; offsets are managed by the +Runtime and committed to its own offset store. All standard `security.*` / `sasl.*` / `ssl.*` +kafka-clients keys are passed through (see `kafka-client.properties` in the runtime `conf/`). + +## 3. Runtime ports & paths + +Usually set via `-D` by `bin/start.sh`; override here if needed. + +| Key | Default | Description | +|---|---|---| +| `eventmesh.http.port` | `8080` | Traffic HTTP (`/events/*`, `/agent/*`, `/session/*`) | +| `eventmesh.admin.port` | `8081` | Admin HTTP (`/admin/*`, `/metrics`) | +| `eventmesh.ws.port` | `-1` (disabled) | WebSocket push port; set e.g. `8082` to enable | +| `eventmesh.offset.path` | `./data/offset` | Local offset store directory | + +## 4. Security + +### TLS / mTLS (transport) + +TLS terminates at the HTTP server. Configure via `UniHttpServer.withTls(SSLContext)` / +`withClientAuth(boolean)` when embedding the runtime, or via the standard JVM system +properties when running from `bin/`. + +### Auth / ACL / quota / audit — the security gate (#5304) + +The runtime ships a unified `SecurityGate` (see +`org.apache.eventmesh.runtime.security.gate`). It is **opt-in**: when no gate is installed, +behavior is unchanged. When installed via `UniHttpServer.withSecurityGate(...)` / +`A2AGatewayHttpHandler.withSecurityGate(...)` / `ConnectorScheduler.withSecurityGate(...)`, +every request flows through one `RequestContext` and is checked in order: + +1. **Authentication + ACL** — the existing filter chain (`TokenAuthFilter`, `AclFilter`, + `SignatureVerifierFilter`, ...) with rules hot-swappable from Meta +2. **Quota** — per-tenant `QuotaManager` (connections / subscriptions / throughput / backlog); + `TenantQuotaManager` is the in-memory default, `QuotaManager.unlimited()` disables quota +3. **Audit** — `AuditSink` SPI; `LoggingAuditSink` (default) emits one structured line per + authorized operation + +Rejections map to HTTP `401` (unauthenticated), `403` (forbidden), `429` (quota exceeded). + +## 5. Rate limiting (per-topic) + +`UniIngressService` enforces an optional per-topic token bucket; configure limits +programmatically via `configureTopicRateLimit(topic, capacity, permitsPerSecond)`. Exhausted +buckets make publish fail with `RateLimitedException` → HTTP `429` on the publish endpoint. +Cluster-wide limits can be inspected and adjusted at `POST /admin/ratelimit`. + +## 6. Admin endpoints (port 8081) + +| Endpoint | Purpose | +|---|---| +| `GET /admin/health` | liveness (also `GET /admin/health` returns `{"status":"UP"}`) | +| `GET /admin/metrics` | JSON runtime metrics | +| `GET /metrics` | Prometheus exposition | +| `GET /admin/subscriptions` | active subscriptions | +| `GET /admin/offsets` | tracked offsets | +| `GET /admin/clients` | connected clients | +| `POST /admin/client/reject` | evict a client | +| `POST /admin/dlq/replay` | replay dead-letter events | +| `GET /admin/dlq/browse` | inspect dead-letter events | +| `GET/POST /admin/ratelimit` | inspect / adjust rate limits | +| `GET/POST /admin/connectors` | connector definitions (CRUD) | +| `GET /admin/connector-workers` | connector worker registry | + +## 7. Traffic endpoints (port 8080) + +Full request/response shapes: see the [client guide](eventmesh-client-guide.md). +Summary: + +| Group | Endpoints | +|---|---| +| Core pub/sub | `/events/publish`, `/events/publish-batch`, `/events/subscribe`, `/events/unsubscribe`, `/events/ack`, `/events/poll` | +| Request-reply | `/events/request`, `/events/reply` | +| Streaming push | `/events/stream` (SSE) | +| Lite topics | `/events/lite/create`, `/events/lite/publish[-bytes]`, `/events/lite/poll[-bytes]` | +| Agent control | `/agent/register`, `/agent/ready`, `/agent/heartbeat`, `/agent/unregister` | +| Sessions | `/session/open`, `/session/recommend`, `/session/close`, `/session/stream`, `/session/publish`, `/session/subscribe` | +| Legacy bridge | `/eventmesh/publish`, `/eventmesh/subscribe`, `/eventmesh/unsubscribe` (old SDK compat) | +| A2A gateway | `/a2a/tasks` family — separate port, [A2A docs](eventmesh-a2a-protocol.md) (Experimental) | + +## 8. Deployment checklist + +- [ ] `EVENTMESH_STORAGE_TYPE` and the backend address set consistently on every instance +- [ ] `eventmesh.offset.path` points at persistent storage (survives restarts) +- [ ] Decide the WebSocket port (default disabled) +- [ ] Production: install a `SecurityGate` (auth tokens + ACL rules + quota + audit) — see §4 +- [ ] Set per-topic rate limits for known hot topics +- [ ] Point monitoring at `/metrics` (Prometheus) and alerts at `/admin/health` diff --git a/docs/eventmesh-features.md b/docs/eventmesh-features.md new file mode 100644 index 0000000000..e570a56bcb --- /dev/null +++ b/docs/eventmesh-features.md @@ -0,0 +1,449 @@ +# Apache EventMesh Features + +> **Status:** Living document. Reflects the post-#5296 architecture-review state +> of the `develop` branch. The **capability status table** in the project README +> is the single source of truth for GA / Beta / Experimental / Legacy tags; this +> page describes *what each feature does* and *where the code lives*. +> +> See [docs/eventmesh-architecture.md](eventmesh-architecture.md) for the +> structural view (control / data / agent planes) and +> [docs/eventmesh-configuration.md](eventmesh-configuration.md) for the keys +> that turn each feature on. + +This page is organized by **user intent**: a user who wants to publish events, +who wants to subscribe, who wants to wire a multi-agent system, who wants to +add a new storage backend, who wants to deploy at scale, and who wants to +operate the cluster. Each section points at the exact code locations you can +read to go deeper. + +--- + +## 1. CloudEvents-native publish / subscribe + +**What it is.** The primary user path. A Java client uses +`CloudEventsClient` to `publish` and `subscribe` over plain HTTP, with the +event body in [CloudEvents 1.0](https://cloudevents.io) format. The runtime +treats the broker as a pure write-ahead log; **all subscription semantics +live in the runtime**, not the broker. + +**Why it matters.** Vendor-neutral events; no client-side knowledge of the MQ; +horizontal scale-out is a config change; reliability is owned by the runtime +through self-managed offsets + explicit ACK. + +**Where the code lives.** + +* Client: `eventmesh-sdks/.../cloudevents/CloudEventsClient.java` +* Walkthrough: [docs/eventmesh-client-guide.md](eventmesh-client-guide.md) +* Runtime HTTP entry: `eventmesh-runtime/.../http/UniHttpServer.java` +* Ingress: `eventmesh-runtime/.../ingress/UniIngressService.java` +* Producer (MQ-as-WAL writer): + `eventmesh-runtime/.../protocol/producer/Producer.java` +* Subscribe: `eventmesh-runtime/.../protocol/subscribe/SubscribeProcessor.java` + +**Configuration highlights.** + +``` +eventmesh.runtime.http.port = 10106 +eventmesh.runtime.http.tls.enabled = true +eventmesh.runtime.subscriptions.maxPerClient = 1000 +``` + +See [docs/eventmesh-configuration.md](eventmesh-configuration.md#publishsubscribe) +for the full list. + +--- + +## 2. Multiple subscriber transports + +**What it is.** A subscriber picks one of three delivery transports at +`POST /events/subscribe` time: + +| Transport | Endpoint | Use case | +| --- | --- | --- | +| HTTP long-polling | `POST /events/subscribe` (default) | Best fit for batch backends, scheduled jobs, serverless | +| Server-Sent Events | `POST /events/subscribeSse` | Browser / mobile push, one-way streaming | +| WebSocket | `POST /events/subscribeWs` | Bi-directional push, low-latency interactive clients | +| Request-reply | `POST /events/request` + `POST /events/reply` | RPC-style synchronous call (built on top of the same MQ-as-WAL pipe) | + +**Why it matters.** Different consumers have different latency / connection +profiles. EventMesh does not force a one-size-fits-all transport; the runtime +fan-out is shared so reliability and quota are uniform. + +**Where the code lives.** + +* Long-poll: `eventmesh-runtime/.../protocol/subscribe/SubscribeProcessor.java` +* SSE: `eventmesh-runtime/.../protocol/subscribe/SseProcessor.java` +* WebSocket: `eventmesh-runtime/.../http/UniWsServer.java` +* Request-reply: `eventmesh-runtime/.../protocol/subscribe/RequestReplyProcessor.java` + +**Configuration highlights.** + +``` +eventmesh.runtime.sse.heartbeatIntervalMs = 15000 +eventmesh.runtime.ws.maxFrameSize = 65536 +eventmesh.runtime.request.reply.timeoutMs = 30000 +``` + +--- + +## 3. Reliable delivery: at-least-once, retries, dead-letter + +**What it is.** Every delivery is tracked by a `DeliveryStateStore`. On ACK +the runtime advances the offset; on failure it retries with backoff; on +terminal failure the event is moved to a `DeadLetterStore`. Operators can +inspect and replay from the dead-letter store. + +**Why it matters.** When the broker is just a WAL, *the runtime* must own +delivery. The storage SPI does not know what is consumed; the runtime +decides when to advance, when to retry, and when to give up. + +**Where the code lives.** + +* State: `eventmesh-runtime/.../state/DeliveryStateStore.java` +* Dead-letter: `eventmesh-runtime/.../state/DeadLetterStore.java` +* Reliable dispatcher: `eventmesh-runtime/.../protocol/producer/ReliableDispatcher.java` +* Retry policy: `eventmesh-common/.../retry/RetryPolicy.java` +* Tests: `eventmesh-runtime/.../state/DeliveryStateStoreTest.java` + +**Configuration highlights.** + +``` +eventmesh.runtime.delivery.maxRetries = 5 +eventmesh.runtime.delivery.backoff.initialMs = 500 +eventmesh.runtime.delivery.backoff.maxMs = 30000 +eventmesh.runtime.deadletter.topic = eventmesh-deadletter +``` + +--- + +## 4. Unified state control plane (issue #5301) + +**What it is.** Before #5301 the runtime had half a dozen overlapping state +APIs (`PartitionOwnership`, `MetaBackedOffsetStore`, `ClusterCoordinator`, +in-memory `TaskRegistry` for A2A). #5301 consolidated them into a small, +federated set of *capability-typed* stores: + +| Level | Interface | Default backend | Purpose | +| --- | --- | --- | --- | +| L1 — local-only | `OffsetStore` | RocksDB or local KV | per-instance delivery hints | +| L2 — cluster-shared | `SubscriptionStore`, `SessionStore` | Meta store (Nacos / Consul / ETCD / ZK) | subscription / session / agent-card registry | +| L3 — durable-egress | `DeadLetterStore`, `TaskStore` | Meta store (CAS + epoch) | dead-letter, A2A task records | + +**Why it matters.** One contract per concern, one TCK per contract, and the +fencing token on the L2/L3 stores makes split-brain impossible to corrupt +state. Adding a new meta backend (e.g. Eureka) is now a single +`MeshStoragePlugin` implementation that passes the TCK. + +**Where the code lives.** + +* Interfaces: `eventmesh-storage-plugin/eventmesh-storage-api/.../storage/MeshStoragePlugin.java` +* TCK: `eventmesh-storage-plugin/eventmesh-storage-api/.../storage/tck/MeshStoragePluginTCK.java` +* Capability flags: `eventmesh-storage-plugin/eventmesh-storage-api/.../storage/StorageCapabilities.java` +* Meta-side adapters: `eventmesh-runtime/.../cluster/ClusterSubscriptionStore.java`, + `eventmesh-runtime/.../session/SessionRegistry.java`, + `eventmesh-runtime/.../a2a/A2AGatewayService.java` + +**Configuration highlights.** + +``` +eventmesh.storage.meta.backend = nacos +eventmesh.storage.meta.nacos.serverAddr = nacos:8848 +eventmesh.storage.local.backend = rocksdb +eventmesh.storage.local.rocksdb.path = /var/lib/eventmesh/offset +``` + +--- + +## 5. Unified security gate (issue #5304) + +**What it is.** An opt-in gate that runs on every ingress point +(`UniHttpServer`, `A2AGatewayHttpHandler`, `ConnectorScheduler`). The gate +composes the existing `FilterChain` (auth + ACL) with a per-tenant +`QuotaManager` and an `AuditSink`. + +**Order of checks.** `FilterChain.invoke` → (on allow) `QuotaManager.acquire` → +`AuditSink.emit`. Rejecting at any step short-circuits downstream and the request +never touches the storage SPI. + +**Why it matters.** A single, composable policy point is the only way to +guarantee that quota / audit / ACL are not forgotten in a new endpoint. The +gate is opt-in so legacy deployments keep working, but new endpoints are +expected to install it. + +**Where the code lives.** + +* Package: `eventmesh-runtime/.../security/gate/` +* Core: `SecurityGate.java`, `RequestContext.java`, `GateDecision.java` +* Quota: `QuotaManager.java`, `UnlimitedQuotaManager.java`, `TenantQuotaManager.java` +* Audit: `AuditSink.java`, `LoggingAuditSink.java`, `DisabledAuditSink.java` +* Wiring: + * `eventmesh-runtime/.../http/UniHttpServer.java#withSecurityGate` + * `eventmesh-runtime/.../a2a/A2AGatewayHttpHandler.java#withSecurityGate` + * `eventmesh-runtime/.../connector/ConnectorScheduler.java#withSecurityGate` +* Tests: `eventmesh-runtime/.../security/gate/SecurityGateTest.java` (10 tests) + +**Configuration highlights.** + +``` +eventmesh.security.gate.enabled = true +eventmesh.security.gate.quota.connections.perTenant = 10000 +eventmesh.security.gate.quota.subscriptions.perTenant = 1000 +eventmesh.security.gate.quota.throughput.bytesPerSec = 1048576 +eventmesh.security.gate.audit.sink = logging +``` + +For the wiring contract and the full list of `RequestContext` fields, see +[docs/eventmesh-architecture.md §4](eventmesh-architecture.md#4-security-gate-issue-5304). + +--- + +## 6. Agent-to-Agent (A2A) protocol + +**What it is.** A2A is the [Agent-to-Agent](eventmesh-a2a-protocol.md) contract — a +task lifecycle (`submitted → working → completed | failed | canceled`) on +top of CloudEvents, with **durable** task records in the meta store. EventMesh +acts as the gateway: it accepts `tasks/send`, dispatches to the target agent +via a topic, and streams replies back over `/a2a/tasks/{taskId}/stream`. + +**Why it matters.** A2A turns EventMesh into an **agent bus**. Synchronous +MCP / JSON-RPC 2.0 tool calls and asynchronous pub/sub share the same +storage, the same quota, the same audit, and the same reliability layer. The +A2A `taskEpoch` field is the cross-agent idempotency key. + +**Where the code lives.** + +* Wire spec: [docs/eventmesh-a2a-protocol.md](eventmesh-a2a-protocol.md) +* Service: `eventmesh-runtime/.../a2a/A2AGatewayService.java` +* Topic convention: `eventmesh-protocol-plugin/eventmesh-protocol-a2a/.../A2ATopicFactory.java` + (`agentInbox(agentId)`, `gatewayResponseTopic(ns, gw, taskId)`) +* Runtime bridge: `eventmesh-runtime/.../a2a/EventMeshA2ATransport.java` +* HTTP handler: `eventmesh-runtime/.../a2a/A2AGatewayHttpHandler.java` +* Server (Netty): `eventmesh-runtime/.../a2a/A2AGatewayServer.java` +* Agent cards: `eventmesh-runtime/.../a2a/AgentCardRegistry.java` + + `InMemoryAgentCardRegistry.java` +* Example agent: `eventmesh-agent/.../Agent.java` + +**Status.** **Experimental.** The wire format and the task lifecycle are +stable; the reaper and the Meta-backed `AgentCardRegistry` are pending (D2 of +issue `#5302`). + +--- + +## 7. Pluggable storage backends + +**What it is.** Every state store (`OffsetStore`, `SubscriptionStore`, …) is +an SPI. Backends currently shipping: + +* **RocketMQ** — `eventmesh-storage-plugin/eventmesh-storage-rocketmq/` +* **Kafka** — `eventmesh-storage-plugin/eventmesh-storage-kafka/` +* **Pulsar** — `eventmesh-storage-plugin/eventmesh-storage-pulsar/` +* **RabbitMQ** — `eventmesh-storage-plugin/eventmesh-storage-rabbitmq/` +* **Redis** — `eventmesh-storage-plugin/eventmesh-storage-redis/` +* **RocksDB** (local) — `eventmesh-storage-plugin/eventmesh-storage-rocksdb/` + (L1, no meta store required) + +**Why it matters.** A new backend (Pravega, AutoMQ, …) is one module that +implements `MeshStoragePlugin` and passes `MeshStoragePluginTCK`. There is no +fork; there is no runtime patch. + +**Where the contract lives.** + +* `eventmesh-storage-plugin/eventmesh-storage-api/.../storage/MeshStoragePlugin.java` +* `eventmesh-storage-plugin/eventmesh-storage-api/.../storage/StorageCapabilities.java` + — capability flags (`OFFSET_TRACKING`, `PREFIX_WATCH`, `CAS`, …) so the + runtime knows what a backend can do without a feature-detect probe. +* `eventmesh-storage-plugin/eventmesh-storage-api/.../storage/tck/MeshStoragePluginTCK.java` + — the test every backend must pass before it can ship. + +**Configuration highlights.** + +``` +eventmesh.storage.meta.backend = nacos +eventmesh.storage.local.backend = rocksdb +eventmesh.storage.rocketmq.namesrvAddr = rocketmq:9876 +eventmesh.storage.kafka.bootstrapServers = kafka:9092 +``` + +--- + +## 8. Connector ecosystem (24+ plugins) + +**What it is.** Connectors copy events between EventMesh and an external +system. The runtime treats them as a first-class concern, but the connector +processes are **separate**: they run in `eventmesh-connector-runtime` and +talk to the data plane over HTTP+CloudEvents. + +Plugins shipping today cover the common categories: + +* **Messaging** — Kafka, RocketMQ, RabbitMQ, Pulsar, Redis +* **Database** — JDBC, MongoDB, MySQL CDC (via Canal) +* **Storage** — S3, file +* **HTTP / API** — HTTP, Knative, OpenFunction, Spring +* **ChatOps** — DingTalk, Slack, WeChat, WeCom, Lark +* **AI** — ChatGPT, MCP (Model Context Protocol) +* **Observability** — Prometheus + +**Where the code lives.** + +* SPI: `eventmesh-connector-plugin/eventmesh-connector-api/` +* Host: `eventmesh-connector-runtime/.../ConnectorManager.java`, + `eventmesh-connector-runtime/.../ConnectorAdminServer.java` +* Per-connector source: `eventmesh-connector-plugin/eventmesh-connector-/` +* Lifecycle: `eventmesh-runtime/.../connector/ConnectorScheduler.java` + +**Configuration highlights.** + +``` +eventmesh.connector.runtime.workerThreads = 16 +eventmesh.connector..class = org.apache.eventmesh.connector.file.FileConnector +eventmesh.connector..mode = source-sink +eventmesh.connector..topic = persistent://public/default/file-events +``` + +**Status.** **Beta** (capability status table). The 24 plugins are +production-usable; the new connector SPI migration (#5288) is still being +rolled out across the plugins. + +--- + +## 9. Pluggable meta service + +**What it is.** The L2/L3 state stores are backed by a meta service. +Backends currently supported: **Consul**, **Nacos**, **ETCD**, **Zookeeper**. +A new meta backend only needs to implement the `MeshStoragePlugin` SPI; the +runtime does not care. + +**Why it matters.** Most enterprise environments already run one of these +for service discovery or config. EventMesh can reuse that cluster, so +operators do not run a new stateful tier just for EventMesh. + +**Where the code lives.** + +* `eventmesh-storage-plugin/eventmesh-storage-nacos/` +* `eventmesh-storage-plugin/eventmesh-storage-consul/` +* `eventmesh-storage-plugin/eventmesh-storage-etcd/` +* `eventmesh-storage-plugin/eventmesh-storage-zookeeper/` +* Watch: `eventmesh-runtime/.../cluster/DynamicConfigWatcher.java` + +--- + +## 10. Filtering, transformation, schema + +**What it is.** A subscriber can attach a **filter expression** at +`POST /events/subscribe` so the runtime only delivers events that match +(Cel-style expression evaluated server-side). A **transformer** can rewrite +the CloudEvent before delivery. Schemas are managed out-of-band by the +[EventMesh-catalog](https://github.com/apache/eventmesh-catalog) project +using AsyncAPI. + +**Where the code lives.** + +* Filter: `eventmesh-runtime/.../protocol/subscribe/filter/` +* Transformer: `eventmesh-runtime/.../protocol/subscribe/transform/` +* Schema (catalog): `eventmesh-catalog/` + +--- + +## 11. Serverless workflow + +**What it is.** EventMesh ships a workflow engine +([EventMesh-workflow](https://github.com/apache/eventmesh-workflow)) that +runs [Serverless Workflow](https://serverlessworkflow.io/) definitions over +events. A workflow can be triggered by an event, listen for follow-up +events, and call back into the runtime to publish more events — closing the +loop on event orchestration. + +--- + +## 12. Observability and operations + +**Metrics.** Prometheus exporter on +`eventmesh.runtime.metrics.port` (default 9090). Pre-built Grafana +dashboards in `eventmesh-examples/observability/`. + +**Logs.** SLF4J; structured JSON layout optional +(`eventmesh.runtime.log.json = true`). + +**Traces.** `RequestContext.traceContext` propagates W3C trace headers +through every plane. OpenTelemetry SDK integration is a single configuration +key. + +**Health & admin.** `/health`, `/metrics`, `/admin/cluster`, and +`/admin/connectors` are exposed by the admin server. See +[docs/eventmesh-configuration.md](eventmesh-configuration.md#admin) +for the full surface. + +**Runbooks.** [docs/production-readiness.md](production-readiness.md) +covers deployment topology, SLOs, and incident response. + +--- + +## 13. Architecture-guard (issue #5305) + +**What it is.** A separate Gradle module, +`eventmesh-architecture-guard`, that hosts ArchUnit rules enforcing the +layered architecture. The rules run **twice**: + +1. locally on `./gradlew :eventmesh-architecture-guard:check` (30-second + feedback loop), and +2. on every PR via `.github/workflows/architecture-guard.yml`. + +A rule violation **fails the build** in both modes. The 30-second local +loop is the key value: it catches layering bugs in the same commit that +introduces them, not eight minutes later in CI. + +**Where the code lives.** + +* Rules: `eventmesh-architecture-guard/.../guard/ArchitectureRules.java` +* Tests: `eventmesh-architecture-guard/.../ArchitectureRulesTest.java` +* CI: `.github/workflows/architecture-guard.yml` + +--- + +## 14. Legacy compatibility (TCP / gRPC / OpenMessaging) + +**What it is.** The original EventMesh wire protocols (TCP + gRPC + +OpenMessaging) still work against the new runtime. The new +`EventMeshFrame` adaptor (`#5299`) wraps the legacy `MeshMessage` / +`OpenMessage` in the same frame type the HTTP path uses, so a single +`FilterChain` and a single `SecurityGate` cover both. + +**Status.** **Legacy-compatible** (capability status table). Existing users +are not broken; new users should use the HTTP + CloudEvents path. + +**Migration.** See +[docs/eventmesh-client-guide.md §1.2](eventmesh-client-guide.md) +for the drop-in replacement of the legacy `EventMeshClient` with +`CloudEventsClient`. + +--- + +## 15. Internationalization + +The runtime and the docs are English-first. The admin server accepts +`Accept-Language` for error messages. The README and most user-facing docs +have Chinese translations under `README.zh-CN.md` and module doc files +with `.zh-CN.md` suffix. + +--- + +## Feature → module map (at a glance) + +| Feature | Module | Key file | +| --- | --- | --- | +| Publish / subscribe | `eventmesh-runtime` | `http/UniHttpServer.java` | +| SSE | `eventmesh-runtime` | `protocol/subscribe/SseProcessor.java` | +| WebSocket | `eventmesh-runtime` | `http/UniWsServer.java` | +| Reliable delivery | `eventmesh-runtime` | `state/DeliveryStateStore.java` | +| State control plane | `eventmesh-runtime` + `eventmesh-storage-plugin` | `cluster/`, `state/`, `session/` | +| Security gate | `eventmesh-runtime` | `security/gate/SecurityGate.java` | +| A2A | `eventmesh-runtime` + `eventmesh-agent` | `a2a/A2AGatewayService.java` | +| Storage backends | `eventmesh-storage-plugin/*` | per-backend `MeshStoragePlugin` impl | +| Meta service | `eventmesh-storage-plugin/*` (Nacos / Consul / ETCD / ZK) | per-backend impl | +| Connectors | `eventmesh-connector-runtime` + `eventmesh-connector-plugin/*` | `ConnectorManager.java` | +| Workflow | `EventMesh-workflow` (separate repo) | — | +| Schema / catalog | `EventMesh-catalog` (separate repo) | — | +| Architecture rules | `eventmesh-architecture-guard` | `guard/ArchitectureRules.java` | +| Client SDKs | `eventmesh-sdks` | per-language package | +| Observability | `eventmesh-common` | `metrics/`, `trace/` | +| Legacy protocols | `eventmesh-runtime` | `protocol/meshmessage/`, `protocol/grpc/` | diff --git a/docs/eventmesh-getting-started.md b/docs/eventmesh-getting-started.md new file mode 100644 index 0000000000..63fa4524ff --- /dev/null +++ b/docs/eventmesh-getting-started.md @@ -0,0 +1,155 @@ +# EventMesh Getting Started + +This guide takes you from zero to a running EventMesh Runtime with a working publisher and +subscriber, using the recommended **HTTP + CloudEvents** path. Configuration reference: +[`eventmesh-configuration.md`](eventmesh-configuration.md). SDK details: +[`eventmesh-client-guide.md`](eventmesh-client-guide.md). + +> Capability maturity levels (GA / Beta / Experimental / Legacy) are defined in the +> [capability status table](../README.md#capability-status) in the main README. + +--- + +## 1. Prerequisites + +- JDK 21+ (Temurin recommended) +- Docker (for the container path), or a local install of one storage backend: + - [Apache RocketMQ](https://rocketmq.apache.org) 4.x or 5.x, **or** + - [Apache Kafka](https://kafka.apache.org) 2.8+ (3.x recommended) +- (SDK only) Java 11+ application with `eventmesh-sdk-java` on the classpath + +## 2. Choose a storage backend + +The EventMesh Runtime is stateless — it owns subscriptions, offsets and delivery, and uses the +MQ purely as a write-ahead log (WAL). Pick one backend per deployment; the client side never +changes. + +| Backend | Type value | Notes | +|---|---|---| +| RocketMQ 4.x | `rocketmq` | classic PULL over remoting | +| RocketMQ 5.x | `rocketmq5` | 5.x POP + Lite Topic support | +| Kafka | `kafka` | assign+seek+poll (no consumer groups), SASL/SSL supported | + +## 3. Run the Runtime + +### Option A — Docker + +```shell +sudo docker pull apache/eventmesh:latest +sudo docker run -d --name eventmesh \ + -e EVENTMESH_STORAGE_TYPE=kafka \ + -e EVENTMESH_KAFKA_NAMESRV=YOUR_KAFKA:9092 \ + -p 8080:8080 -p 8081:8081 \ + apache/eventmesh:latest +``` + +Ports: `8080` = traffic HTTP (`/events/*`), `8081` = admin HTTP (`/admin/*`). The WebSocket +push port (`8082`) and the connector runtime admin port (`8083`) are opt-in. + +### Option B — From source + +```shell +git clone https://github.com/apache/eventmesh.git +cd eventmesh + +# pick your backend via EVENTMESH_STORAGE_TYPE (rocketmq | rocketmq5 | kafka) +export EVENTMESH_STORAGE_TYPE=kafka +export EVENTMESH_KAFKA_NAMESRV=localhost:9092 + +./gradlew :eventmesh-runtime:clean :eventmesh-runtime:dist +cd eventmesh-runtime/dist && bash bin/start.sh +``` + +Storage-specific keys (all overridable via `-D` system properties) are documented in +[`eventmesh-configuration.md`](eventmesh-configuration.md#storage-backends). + +### Verify it is up + +```shell +curl http://localhost:8081/admin/health +# {"status":"UP"} +``` + +## 4. Publish your first event + +Applications send standard [CloudEvents](https://cloudevents.io) 1.0 over HTTP. `202 Accepted` +means the event is durably in the WAL: + +```shell +curl -X POST "http://localhost:8080/events/publish?topic=orders" \ + -H "Content-Type: application/cloudevents+json" \ + -d '{ + "specversion": "1.0", + "id": "89010a5a-3c6f-4a1e-9b2d-0f7c1f2e3a4b", + "source": "/example/producer", + "type": "com.example.order.created", + "datacontenttype": "application/json", + "data": {"orderId": 42, "amount": 99.5} + }' +``` + +## 5. Subscribe and receive + +Register a subscription (there are **no consumer groups** — EventMesh tracks offsets itself), +then receive via one of three transports: + +```shell +# 1. register: clientId + topic + distribution mode +curl -X POST http://localhost:8080/events/subscribe \ + -H "Content-Type: application/json" \ + -d '{"clientId":"order-svc","topic":"orders","mode":"LOAD_BALANCE"}' + +# 2a. HTTP long-polling +curl "http://localhost:8080/events/poll?clientId=order-svc&topics=orders&timeout=30000" + +# 2b. after processing, acknowledge so the offset advances (at-least-once) +curl -X POST http://localhost:8080/events/ack \ + -H "Content-Type: application/json" \ + -d '{"clientId":"order-svc","deliveryIds":["..."]}' +``` + +Distribution modes: + +| Mode | Semantics | +|---|---| +| `LOAD_BALANCE` | one subscriber among the group receives each event (partition-key sticky variant available) | +| `BROADCAST` | every subscriber receives every event | +| `MULTICAST` | subscriber-side predicate filters events per client | + +SSE and WebSocket push are also available (`GET /events/stream`, `subscribeWs` in the SDK) — +see the client guide for the trade-offs. + +## 6. Use the SDK instead of raw HTTP (recommended) + +```java +CloudEventsClient client = CloudEventsClient.builder() + .baseUrl("http://localhost:8080") + .build(); +client.init(); + +// publish +client.publish("orders", CloudEventBuilder.v1() + .withId(UUID.randomUUID().toString()) + .withSource(URI.create("/order-svc")) + .withType("com.example.order.created") + .withData("application/json", "{\"orderId\":42}".getBytes(UTF_8)) + .build()); + +// subscribe — handler return implies auto-ACK +client.subscribe("orders", "LOAD_BALANCE", event -> { + System.out.println("got " + event.getType()); +}); +``` + +Full API (request/reply, streaming sessions, lite topics, SSE/WS): see the +[client guide](eventmesh-client-guide.md). + +## 7. Where to go next + +- [Configuration reference](eventmesh-configuration.md) — every runtime key, per-backend settings +- [Client guide](eventmesh-client-guide.md) — complete SDK walkthrough +- [Production readiness](production-readiness.md) — verified capabilities, SLOs, runbooks +- [A2A gateway](eventmesh-a2a-protocol.md) — agent-to-agent messaging (Experimental) +- Admin API (`/admin/*`) quick reference: `metrics`, `subscriptions`, `offsets`, `clients`, + `client/reject`, `dlq/replay`, `dlq/browse`, `ratelimit`, `health`, `connectors`, + `connector-workers` on port 8081 diff --git a/docs/eventmesh-offset-lb-frame-design.md b/docs/eventmesh-offset-lb-frame-design.md index 7441d91924..70adce1ad7 100644 --- a/docs/eventmesh-offset-lb-frame-design.md +++ b/docs/eventmesh-offset-lb-frame-design.md @@ -1,5 +1,8 @@ # EventMesh Offset 管理 · 负载均衡 · Frame 协议转换 设计总结 +> **状态说明**：本文是 offset/负载均衡/Frame 的设计文档（历史记录）。各能力的当前实现状态 +> 以主 README 的[能力状态表](../README.md#能力状态capability-status)为准。 + > 整合 offset 管理、全面粘性负载均衡、EventMeshFrame 协议转换三块设计的完整方案（2026-08-13）。 > 关联文档：[`eventmesh-uni-architecture-redesign.md`](./eventmesh-uni-architecture-redesign.md) §19 架构深化。 diff --git a/docs/eventmesh-uni-architecture-redesign.md b/docs/eventmesh-uni-architecture-redesign.md index 25460b09ac..09ed20d81c 100644 --- a/docs/eventmesh-uni-architecture-redesign.md +++ b/docs/eventmesh-uni-architecture-redesign.md @@ -1,4 +1,7 @@ # EventMesh 架构简化重构方案 + +> **状态说明**：本文是架构重写的设计文档（历史记录）。各能力的当前实现状态与成熟度 +> 以主 README 的[能力状态表](../README.md#能力状态capability-status)为准。 > 分支：`refactor/unified-runtime-pipeline` → 基于 本次讨论的全新方向 > diff --git a/docs/production-readiness.md b/docs/production-readiness.md index fff0b446b6..b917450655 100644 --- a/docs/production-readiness.md +++ b/docs/production-readiness.md @@ -9,6 +9,10 @@ **MQ-as-stateless-WAL + HTTP SDK（CloudEvents/MeshMessage/A2A 多协议）+ EventMesh 自管订阅/offset + 内部 EventMeshFrame** 的重写（详见 `docs/eventmesh-uni-architecture-redesign.md`）。采用叠加式策略：新核心并行存在，旧 TCP/HTTP 保留为兼容适配层（老客户端零改动）。 +> **能力状态定级**：各能力（HTTP+CloudEvents / 存储 / SSE·WS / Connector / A2A / Legacy SDK） +> 的状态、建议与迁移目标以主 README 的[能力状态表](../README.md#能力状态capability-status)为准； +> 本文档记录生产准入的验证与运维细节。 + ## 2. 模块（活跃 gradle） | 类别 | 模块 |