[ISSUE #5411] Serve the legacy SDK gRPC protocol on the v2 runtime (port 10205 bridge) + connector plugin tests 23/23 (#5412)
* [ISSUE #5411] Serve the legacy SDK gRPC protocol on the v2 runtime (port 10205 bridge)
EventMeshGrpcServer binds PublisherService/ConsumerService/HeartbeatService on
eventmesh.grpc.port (opt-in, 0/-1 = off) and maps every call onto the v2
UniIngressService pipeline (WAL at-least-once, shared retry/DLQ):
- publish/batchPublish/publishOneWay/batchPublishOneWay -> ingress.publish (topic = subject)
- requestReply -> ingress.request with the legacy TTL attribute
- webhook subscribe -> WebHookChannel target; CLUSTERING->LOAD_BALANCE, BROADCASTING->BROADCAST
- subscribeStream (bidi) -> GrpcStreamChannel push target, ACKs ride back on the stream
- heartbeat -> GrpcClientRegistry TTL refresh + reaper unsubscribes stale clients
GrpcCloudEventMapper holds the proto<->v2 CloudEvent mapping + legacy Response envelope.
ruleGrpcProtocolHidden grants the sanctioned runtime.grpc.. adapter (TCP-rule precedent).
* [ISSUE #5411] gRPC bridge tests: hermetic mapping unit tests + real-SDK integration
GrpcCloudEventMapperTest covers every proto attribute flavor + the 3-key response
envelope. GrpcLegacyBridgeIntegrationTest boots the bridge over the in-memory
storage and drives it with the real EventMeshGrpcProducer/Consumer (the SDK as its
own conformance suite): publish, batch publish, stream subscription round-trip.
* Add unit tests for the remaining 19 connector plugins (23/23 coverage)
Hermetic where possible: webhook sinks (dingtalk/http/knative/lark/slack/wechat/
wecom/chatgpt) verified against a local capturing HTTP server; mcp (JSON-RPC
envelope), openfunction (Ce-* context headers), prometheus (single merged push),
spring (EventForwarder contract) exercised with their exact wire contracts;
webhook sources (10 plugins) POST their native callback payloads through the
lazy-bound hook port; prometheus/spring sources cover scrape + buffer drains;
external-client plugins (canal/jdbc/mongodb/rabbitmq/redis/s3/pravega) get
contract-level tests (init config-parse where lazy, commit no-op) since their
backends cannot boot in CI. Closes the "4 of 23 plugins carry the only unit
tests" gap tracked under the #5296 review.
* [ISSUE #5411] docs: legacy gRPC bridge served (opt-in); Connector Runtime test status
configuration.md flips eventmesh.grpc.port from RESERVED to served (opt-in, 1.x
default 10205); protocols.md gains the legacy-gRPC mapping table section; both
READMEs note the served bridge and update the Connector Runtime row to 23/23
plugin test coverage (drops the stale "4 of 23" wording).
* docs: make the Connector Runtime GA-criteria cell concrete
Replace the vague "GA criteria tracked under the #5296 architecture review"
wording with the actual remaining GA blockers: per-plugin real-backend
integration tests and the connector-runtime HA story. Also reference #5412
for the 23/23 unit-test coverage.
* docs: refresh remaining cross-references for the #5411 bridge and 23/23 tests
- deployment.md: connector-runtime status drops the stale "only 4 of 23
plugins carry unit tests" (now 23/23 since #5412; real-backend ITs stay
the GA gate)
- architecture/overview.md: protocol-status bullet + legacy-SDK paragraph
mention the served opt-in gRPC bridge; Key-classes table gains the
EventMeshGrpcServer entry
- quickstart/getting-started.md: port list mentions the opt-in 10205
legacy gRPC bridge
diff --git a/README.md b/README.md
index 7efd4d1..bf6dd66 100644
--- a/README.md
+++ b/README.md
@@ -67,7 +67,7 @@
| [Kafka / RocketMQ storage](eventmesh-storage-plugin/) (4.x, 5.x) | **GA target** | Recommended — pluggable WAL backends, TCK-covered (`MeshStoragePluginTCK`) | Primary path |
| [Memory storage](eventmesh-storage-plugin/eventmesh-storage-memory/) (default) | **Beta** | Zero-dependency dev/CI/quick-start backend (`docker run apache/eventmesh` with no broker); state is process-local — not for production | Switch `EVENTMESH_STORAGE_TYPE` to kafka / rocketmq / rocketmq5 |
| SSE / WebSocket push | **Beta** | Usable — integration-tested; ACK-tracked redelivery + DLQ are shared with long-polling (same `ReliableDispatcher`), e2e-real-broker suite in [#5389](https://github.com/apache/eventmesh/pull/5389) | Unified push transports |
-| Connector Runtime | **Beta** | Usable — working end-to-end; all 23 plugins fully implemented since [#5394](https://github.com/apache/eventmesh/pull/5394) (no template stubs left), data-loss hardening in [#5328](https://github.com/apache/eventmesh/pull/5328); 4 of 23 (file/kafka/pulsar/rocketmq) still carry the only unit tests | Remaining plugin tests + GA criteria tracked under the #5296 architecture review |
+| Connector Runtime | **Beta** | Usable — working end-to-end; all 23 plugins fully implemented since [#5394](https://github.com/apache/eventmesh/pull/5394) (no template stubs left), data-loss hardening in [#5328](https://github.com/apache/eventmesh/pull/5328); unit tests cover all 23 plugins since [#5412](https://github.com/apache/eventmesh/pull/5412) | Promote to GA once the remaining #5296 review items land (real-backend integration tests per plugin, connector-runtime HA story) |
| [A2A / Agent Gateway](docs/feature/a2a.md) | **Beta** | Usable — task store + runtime bridge (#5302/#5304), task reaper + Meta-backed agent cards in [#5346](https://github.com/apache/eventmesh/pull/5346), quota classification in [#5373](https://github.com/apache/eventmesh/pull/5373); the Testcontainers E2E gate (#5340) closed in September 2026 | Unified Runtime A2A |
| [Agent tools & event triggers](docs/feature/agent-tools.md) | **Experimental** | Evaluate — `LlmClient`/`ConversationMemory`/`AgentTool` extension points, connector-backed tools and SPI plugin deployment landed in [#5408](https://github.com/apache/eventmesh/pull/5408)/[#5409](https://github.com/apache/eventmesh/pull/5409); runtime in `eventmesh-agent-runtime`, plugins in `eventmesh-agent-plugin/` | Agent tool ecosystem |
| TCP / gRPC / OpenMessaging SDKs | **Legacy-compatible** | Existing users only — kept so old clients run unmodified; not extended | [HTTP + CloudEvents](docs/feature/client-java.md) |
@@ -81,7 +81,10 @@
> Migrating off TCP / gRPC SDKs? The legacy clients keep working against the current
> runtime; see the [client guide](docs/feature/client-java.md) for the
-> HTTP + CloudEvents replacement (`CloudEventsClient`).
+> HTTP + CloudEvents replacement (`CloudEventsClient`). The legacy gRPC SDK
+> surface is served again since the #5411 bridge (opt-in via
+> `eventmesh.grpc.port=10205`) — old `EventMeshGrpcProducer` /
+> `EventMeshGrpcConsumer` clients run unmodified on the v2 runtime.
### Documentation
diff --git a/README.zh-CN.md b/README.zh-CN.md
index 15e498f..b2da17d 100644
--- a/README.zh-CN.md
+++ b/README.zh-CN.md
@@ -67,7 +67,7 @@
| [Kafka / RocketMQ 存储](eventmesh-storage-plugin/)（4.x、5.x） | **GA 目标** | 推荐——可插拔 WAL 后端，TCK 覆盖（`MeshStoragePluginTCK`） | 主路径 |
| [Memory 内存存储](eventmesh-storage-plugin/eventmesh-storage-memory/)（默认） | **Beta** | 零依赖的开发/CI/快速上手后端（`docker run apache/eventmesh` 无需 broker）；状态仅存于进程内——不可用于生产 | 切换 `EVENTMESH_STORAGE_TYPE` 为 kafka / rocketmq / rocketmq5 |
| SSE / WebSocket 推送 | **Beta** | 可用——已有集成测试；ACK 追踪的重投递与 DLQ 已与长轮询共享同一 `ReliableDispatcher`，真实 broker 的 e2e 套件见 [#5389](https://github.com/apache/eventmesh/pull/5389) | 统一推送传输 |
-| Connector Runtime | **Beta** | 可用——端到端工作；自 [#5394](https://github.com/apache/eventmesh/pull/5394) 起 23 个插件全部完整实现（不再有模板桩），数据丢失加固见 [#5328](https://github.com/apache/eventmesh/pull/5328)；单元测试仍仅覆盖其中 4 个（file/kafka/pulsar/rocketmq） | 剩余插件测试与 GA 标准由 #5296 架构 review 跟踪 |
+| Connector Runtime | **Beta** | 可用——端到端工作；自 [#5394](https://github.com/apache/eventmesh/pull/5394) 起 23 个插件全部完整实现（不再有模板桩），数据丢失加固见 [#5328](https://github.com/apache/eventmesh/pull/5328)；自 [#5412](https://github.com/apache/eventmesh/pull/5412) 起单元测试覆盖全部 23 个插件 | 待 #5296 review 剩余项落地后升级 GA（各插件真实后端集成测试、connector-runtime HA 方案） |
| [A2A / Agent 网关](docs/feature/a2a.md) | **Beta** | 可用——TaskStore + Runtime 桥（#5302/#5304）、任务 reaper 与 Meta 化 AgentCard（[#5346](https://github.com/apache/eventmesh/pull/5346)）、配额分类（[#5373](https://github.com/apache/eventmesh/pull/5373)）均已落地；Testcontainers E2E 门槛（#5340）已于 2026 年 9 月关闭 | 统一 Runtime A2A |
| [智能体工具与事件触发](docs/feature/agent-tools.md) | **实验性** | 评估——`LlmClient`/`ConversationMemory`/`AgentTool` 扩展点、connector 工具化与 SPI 插件化部署已落地（[#5408](https://github.com/apache/eventmesh/pull/5408)/[#5409](https://github.com/apache/eventmesh/pull/5409)）；运行时在 `eventmesh-agent-runtime`，插件在 `eventmesh-agent-plugin/` | 智能体工具生态 |
| TCP / gRPC / OpenMessaging SDK | **Legacy 兼容** | 仅存量用户——保持老客户端零改动运行；不再扩展 | [HTTP + CloudEvents](docs/feature/client-java.md) |
@@ -80,7 +80,11 @@
- **Legacy 兼容** —— 仅为存量客户端零改动兼容而维护；只修缺陷、不加功能。新接入不要选这里。
> 正在从 TCP / gRPC SDK 迁移？Legacy 客户端在当前 Runtime 上继续可用；替代方案（HTTP +
-> CloudEvents 的 `CloudEventsClient`）见[客户端指引](docs/feature/client-java.md)。
+> CloudEvents 的 `CloudEventsClient`）
+
+> #5411 之后 legacy gRPC SDK 接入面重新可用（通过 `eventmesh.grpc.port=10205`
+> 显式开启）——旧的 `EventMeshGrpcProducer` / `EventMeshGrpcConsumer` 客户端
+> 可以零改动运行在 v2 Runtime 上。见[客户端指引](docs/feature/client-java.md)。
### 文档导航
diff --git a/docs/architecture/overview.md b/docs/architecture/overview.md
index 16a0128..e37bb50 100644
--- a/docs/architecture/overview.md
+++ b/docs/architecture/overview.md
@@ -114,7 +114,8 @@
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.
+semantics of the old `MeshMessage` / `OpenMessage` clients; the legacy gRPC
+SDK family is served by the opt-in `EventMeshGrpcServer` bridge (#5411).
### Key classes
@@ -122,6 +123,7 @@
| --- | --- | --- |
| 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 |
+| Legacy gRPC bridge | `eventmesh-runtime/.../grpc/EventMeshGrpcServer.java` | Opt-in port-10205 bridge serving the 1.x `PublisherService`/`ConsumerService`/`HeartbeatService` (#5411) |
| 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 |
@@ -418,7 +420,9 @@
HTTP codec are still supported for backward compatibility but
are explicitly marked Legacy and receive only critical bug
fixes. See `docs/feature/protocols.md` for the migration path.
-* gRPC framing is **Beta**: stable, but the API surface may shift.
+* gRPC framing is **Beta**: stable, but the API surface may shift. The
+ legacy SDK gRPC surface is served again as an opt-in compatibility bridge
+ on `eventmesh.grpc.port` since #5411 (see `docs/feature/protocols.md` §1.1).
* A2A is **Experimental** until the readiness checks in #5340
pass.
diff --git a/docs/feature/deployment.md b/docs/feature/deployment.md
index 1e364e2..344abda 100644
--- a/docs/feature/deployment.md
+++ b/docs/feature/deployment.md
@@ -120,8 +120,10 @@
scheduled by the runtime's `ConnectorScheduler` with **generation
fencing** (#5382): every (re)assignment bumps a per-connector generation,
and a stale delayed start can never take over a connector running a newer
-generation. Status: **Experimental** — only 4 of 23 plugins carry unit
-tests today.
+generation. Status: **Experimental** — all 23 plugins carry unit tests
+since [#5412](https://github.com/apache/eventmesh/pull/5412) (hermetic
+webhook tests + contract tests for external-client plugins); real-backend
+integration tests remain the GA gate.
## Production checklist
diff --git a/docs/feature/protocols.md b/docs/feature/protocols.md
index 20fb1c7..b880520 100644
--- a/docs/feature/protocols.md
+++ b/docs/feature/protocols.md
@@ -20,7 +20,7 @@
| EventMeshFrame (internal) | **GA** | `org.apache.eventmesh.common.wire.EventMeshFrame` (Frame architecture) | runtime internal; producer / storage / push path | - |
| A2A (Agent-to-Agent) | **Experimental** | A2A JSON-RPC + SSE | `eventmesh-protocol-plugin/eventmesh-protocol-a2a`, A2A gateway on Runtime | - |
| MeshMessage TCP | **Legacy** | length-prefixed `MeshMessage` bytes | runtime `tcp/` subpackage, `eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/resolver/tcp` | CloudEvents HTTP, or A2A (for agent workloads) |
-| gRPC (CloudEvents + EventMeshMessage) | **Beta** | protobuf over HTTP/2 | runtime `transport/grpc`, `eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/resolver/grpc` | CloudEvents HTTP for new clients |
+| gRPC (CloudEvents + EventMeshMessage) | **Beta** | protobuf over HTTP/2 | runtime `grpc/` bridge (issue #5411), `eventmesh-protocol-plugin/eventmesh-protocol-meshmessage/resolver/grpc` | CloudEvents HTTP for new clients |
| OpenMessaging API (TCP) | **Legacy** | OMA spec, used by the legacy TCP client only | `eventmesh-sdks/eventmesh-sdk-java/client/tcp/impl/openmessage` | CloudEvents HTTP client |
> **GA** = production-ready and the recommended path. **Beta** = stable but
@@ -28,6 +28,33 @@
> change without notice. **Legacy** = still works, but is no longer the
> recommended choice and is being phased out.
+### 1.1 Legacy gRPC bridge (served, opt-in — issue #5411)
+
+The v2 runtime serves the legacy SDK gRPC protocol as a **compatibility
+bridge** on `eventmesh.grpc.port` (1.x default `10205`; unset / `-1` =
+disabled, opt-in like the WS port). There is no second messaging engine:
+every call maps onto the same v2 ingress/delivery pipeline the HTTP plane
+uses (WAL durability, at-least-once, shared retry/DLQ).
+
+| Legacy call | v2 mapping |
+| --- | --- |
+| `publish` / `batchPublish` / `publishOneWay` / `batchPublishOneWay` | proto `CloudEvent` → v2 CloudEvent, persisted via `UniIngressService.publish` (topic = proto `subject` attribute) |
+| `requestReply` | v2 request/reply correlation (`UniIngressService.request`), TTL attribute drives the timeout |
+| `subscribe` (webhook `url`) | `WebHookChannel` push target + v2 subscription per topic |
+| `subscribeStream` (bidi) | `GrpcStreamChannel` push target pumping the v2 dispatcher into the stream; ACKs ride back as stream replies |
+| `unsubscribe` | v2 unsubscribe per topic + client deregistration |
+| `heartbeat` | TTL refresh in the `GrpcClientRegistry`; a reaper evicts stale clients (unsubscribes them) |
+
+Group semantics: the legacy `consumerGroup` maps onto a v2 subscription
+group; the SDK's default `CLUSTERING` mode maps to `LOAD_BALANCE`
+distribution, `BROADCASTING` maps to `BROADCAST`. The clientId is derived
+from `consumerGroup` + `env` + `idc` (the 1.x triple).
+
+Non-goals (follow-ups): a new gRPC-native v2 API (HTTP + CloudEvents stays
+the primary path), the gRPC admin surface (stays unimplemented as on 1.x),
+and Go/Rust SDK verification (the protos are shared; Java SDK is the
+conformance suite — see `GrpcLegacyBridgeIntegrationTest`).
+
## 2. Server-side protocol plugins
The runtime discovers protocol adaptors via the
diff --git a/docs/quickstart/configuration.md b/docs/quickstart/configuration.md
index 4e5f017..796f879 100644
--- a/docs/quickstart/configuration.md
+++ b/docs/quickstart/configuration.md
@@ -78,7 +78,7 @@
| `eventmesh.http.port` | `10105` | Traffic HTTP (`/events/*`, `/agent/*`, `/session/*`) |
| `eventmesh.admin.port` | `10106` | Admin HTTP (`/admin/*`, `/metrics`) |
| `eventmesh.ws.port` | `-1` (disabled) | WebSocket push port; set e.g. `10107` to enable |
-| `eventmesh.grpc.port` | `10205` | RESERVED for the future gRPC protocol (not served yet; keeps the 1.x default warm) |
+| `eventmesh.grpc.port` | `-1` (disabled) | Legacy SDK gRPC bridge (Publisher/Consumer/Heartbeat services, issue #5411); set `10205` (the 1.x default) to serve old `EventMeshGrpcProducer`/`EventMeshGrpcConsumer` clients on the v2 runtime |
| `eventmesh.a2a.port` | `10108` | A2A gateway REST plane (opt-in via `eventmesh.a2a.enabled=true`) |
| `eventmesh.offset.path` | `./data/offset` | Local offset store directory |
diff --git a/docs/quickstart/getting-started.md b/docs/quickstart/getting-started.md
index 9413b20..76d3886 100644
--- a/docs/quickstart/getting-started.md
+++ b/docs/quickstart/getting-started.md
@@ -54,7 +54,8 @@
backend address keys shown below.
Ports: `10105` = traffic HTTP (`/events/*`), `10106` = admin HTTP (`/admin/*`). The WebSocket
-push port (`10107`) is opt-in.
+push port (`10107`) is opt-in, and the legacy SDK gRPC bridge (`10205`, for old
+`EventMeshGrpcProducer`/`EventMeshGrpcConsumer` clients) is opt-in since #5411.
### Option B — From source
diff --git a/eventmesh-architecture-guard/src/main/java/org/apache/eventmesh/architecture/guard/ArchitectureRules.java b/eventmesh-architecture-guard/src/main/java/org/apache/eventmesh/architecture/guard/ArchitectureRules.java
index 2968b99..53f9657 100644
--- a/eventmesh-architecture-guard/src/main/java/org/apache/eventmesh/architecture/guard/ArchitectureRules.java
+++ b/eventmesh-architecture-guard/src/main/java/org/apache/eventmesh/architecture/guard/ArchitectureRules.java
@@ -65,6 +65,9 @@
.and().resideOutsideOfPackage("org.apache.eventmesh.common..")
.and().resideOutsideOfPackage("org.apache.eventmesh.protocol.meshmessage..")
.and().resideOutsideOfPackage("org.apache.eventmesh.client..")
+ // #5411: the legacy gRPC bridge (runtime.grpc..) is the sanctioned server-side
+ // adapter for these types — same carve-out the TCP rule grants runtime..
+ .and().resideOutsideOfPackage("org.apache.eventmesh.runtime.grpc..")
.should().dependOnClassesThat().resideInAPackage("org.apache.eventmesh.common.protocol.grpc..");
public static ArchRule ruleTcpProtocolHidden = noClasses()
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/package-info.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/package-info.java
index 210f7e6..bc4672a 100644
--- a/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/package-info.java
+++ b/eventmesh-common/src/main/java/org/apache/eventmesh/common/protocol/grpc/package-info.java
@@ -31,8 +31,9 @@
* <p><b>Forbidden (this package must NOT be imported by):</b>
* <ul>
- * <li>{@code ANY module other than eventmesh-sdks:eventmesh-sdk-java and}
- * {@code eventmesh-protocol-plugin:eventmesh-protocol-meshmessage (ruleGrpcProtocolHidden)}</li>
+ * <li>{@code ANY module other than eventmesh-sdks:eventmesh-sdk-java,}
+ * {@code eventmesh-protocol-plugin:eventmesh-protocol-meshmessage and the runtime gRPC bridge}
+ * {@code (org.apache.eventmesh.runtime.grpc.., issue #5411) (ruleGrpcProtocolHidden)}</li>
* </ul>
*/
package org.apache.eventmesh.common.protocol.grpc;
diff --git a/eventmesh-connector-plugin/eventmesh-connector-canal/src/test/java/org/apache/eventmesh/connector/canal/sink/CanalSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-canal/src/test/java/org/apache/eventmesh/connector/canal/sink/CanalSinkConnectorTest.java
new file mode 100644
index 0000000..4ec4583
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-canal/src/test/java/org/apache/eventmesh/connector/canal/sink/CanalSinkConnectorTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.canal.sink;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for CanalSinkConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class CanalSinkConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ CanalSinkConnector connector = new CanalSinkConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(Collections.singletonList(last)); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-canal/src/test/java/org/apache/eventmesh/connector/canal/source/CanalSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-canal/src/test/java/org/apache/eventmesh/connector/canal/source/CanalSourceConnectorTest.java
new file mode 100644
index 0000000..1f71384
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-canal/src/test/java/org/apache/eventmesh/connector/canal/source/CanalSourceConnectorTest.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.canal.source;
+
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for CanalSourceConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class CanalSourceConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ CanalSourceConnector connector = new CanalSourceConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(last); // must not throw
+ }
+
+ @Test
+ void initParsesDocumentedDefaults() {
+ CanalSourceConnector connector = new CanalSourceConnector();
+ connector.init(new Properties()); // must not connect anything
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/test/java/org/apache/eventmesh/connector/chatgpt/sink/ChatgptSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/test/java/org/apache/eventmesh/connector/chatgpt/sink/ChatgptSinkConnectorTest.java
new file mode 100644
index 0000000..62ce4cb
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/test/java/org/apache/eventmesh/connector/chatgpt/sink/ChatgptSinkConnectorTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.chatgpt.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {class} POSTs every event's data bytes to the configured URL and swallows
+ * individual failures (log-and-continue contract). A local HTTP server captures the requests.
+ */
+class ChatgptSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private ChatgptSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(200, 0);
+ ex.close();
+ });
+ server.start();
+ sink = new ChatgptSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.webhookUrl", "http://127.0.0.1:" + server.getAddress().getPort() + "/hook");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putPostsEveryEventData() {
+ sink.put(List.of(event("e1", "alpha"), event("e2", "beta")));
+ assertEquals(List.of("alpha", "beta"), bodies);
+ }
+
+ @Test
+ void putWithEmptyBatchIsNoOp() {
+ sink.put(Collections.emptyList());
+ assertEquals(Collections.emptyList(), bodies);
+ }
+
+ @Test
+ void putWithNullDataPostsEmptyBody() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("n1").withSource(java.net.URI.create("/test"))
+ .withType("test.event").build()));
+ assertEquals(1, bodies.size());
+ assertEquals("", bodies.get(0));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/test/java/org/apache/eventmesh/connector/chatgpt/source/ChatgptSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/test/java/org/apache/eventmesh/connector/chatgpt/source/ChatgptSourceConnectorTest.java
new file mode 100644
index 0000000..b2a835c
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-chatgpt/src/test/java/org/apache/eventmesh/connector/chatgpt/source/ChatgptSourceConnectorTest.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.chatgpt.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: ChatgptSourceConnector receives a POST on its webhook endpoint and poll() surfaces the body
+ * as a CloudEvent (POST -> buffered -> poll contract).
+ */
+class ChatgptSourceConnectorTest {
+
+ private ChatgptSourceConnector source;
+ private HttpServer probe; // keeps an unrelated port warm so stop ordering is deterministic
+
+ @BeforeEach
+ void boot() throws Exception {
+ source = new ChatgptSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/chatgpt");
+ source.init(props);
+ probe = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ probe.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ probe.stop(0);
+ }
+
+ private int hookPort() throws Exception {
+ // the connector binds its port lazily on first poll; trigger it then discover the bound port
+ java.lang.reflect.Field f = ChatgptSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ com.sun.net.httpserver.HttpServer s =
+ (com.sun.net.httpserver.HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ @Test
+ void postedBodyIsPolledAsEvent() throws Exception {
+ source.poll(); // trigger lazy server start
+ int port = hookPort();
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/chatgpt").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write("{\"prompt\":\"hi\"}".getBytes(StandardCharsets.UTF_8));
+ }
+ assertEquals(200, conn.getResponseCode());
+ conn.disconnect();
+
+ java.util.List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "posted body must surface as exactly one event");
+ // the connector answers the prompt: without an OpenAI key the answer is empty
+ assertEquals("{\"prompt\":\"hi\",\"answer\":\"\"}",
+ new String(events.get(0).getData().toBytes(), StandardCharsets.UTF_8));
+ assertTrue(events.get(0).getId() != null && !events.get(0).getId().isEmpty());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/test/java/org/apache/eventmesh/connector/dingtalk/sink/DingtalkSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/test/java/org/apache/eventmesh/connector/dingtalk/sink/DingtalkSinkConnectorTest.java
new file mode 100644
index 0000000..99b1784
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/test/java/org/apache/eventmesh/connector/dingtalk/sink/DingtalkSinkConnectorTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.dingtalk.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {class} POSTs every event's data bytes to the configured URL and swallows
+ * individual failures (log-and-continue contract). A local HTTP server captures the requests.
+ */
+class DingtalkSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private DingtalkSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(200, 0);
+ ex.close();
+ });
+ server.start();
+ sink = new DingtalkSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.webhookUrl", "http://127.0.0.1:" + server.getAddress().getPort() + "/hook");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putPostsEveryEventData() {
+ sink.put(List.of(event("e1", "alpha"), event("e2", "beta")));
+ assertEquals(List.of("alpha", "beta"), bodies);
+ }
+
+ @Test
+ void putWithEmptyBatchIsNoOp() {
+ sink.put(Collections.emptyList());
+ assertEquals(Collections.emptyList(), bodies);
+ }
+
+ @Test
+ void putWithNullDataPostsEmptyBody() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("n1").withSource(java.net.URI.create("/test"))
+ .withType("test.event").build()));
+ assertEquals(1, bodies.size());
+ assertEquals("", bodies.get(0));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/test/java/org/apache/eventmesh/connector/dingtalk/source/DingtalkSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/test/java/org/apache/eventmesh/connector/dingtalk/source/DingtalkSourceConnectorTest.java
new file mode 100644
index 0000000..071134a
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-dingtalk/src/test/java/org/apache/eventmesh/connector/dingtalk/source/DingtalkSourceConnectorTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.dingtalk.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: DingtalkSourceConnector lazily boots its webhook endpoint on first poll; a native dingtalk
+ * callback payload POSTed to the hook surfaces via poll() as a CloudEvent carrying the platform
+ * message id / subject mapping.
+ */
+class DingtalkSourceConnectorTest {
+
+ private static final String BODY = "{\"msgId\":\"m-1\",\"conversationId\":\"c-1\",\"text\":\"hello\"}";
+
+ private int hookPort(DingtalkSourceConnector source) throws Exception {
+ Field f = DingtalkSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ HttpServer s = (HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ private DingtalkSourceConnector boot() {
+ DingtalkSourceConnector source = new DingtalkSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/dingtalk");
+ // no secret/token set -> signature checks are disabled (dev mode)
+ source.init(props);
+ return source;
+ }
+
+ @Test
+ void postedCallbackSurfacesAsEvent() throws Exception {
+ DingtalkSourceConnector source = boot();
+ source.poll(); // trigger lazy server start
+ int port = hookPort(source);
+
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/dingtalk").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setRequestProperty("Content-Type",
+ BODY.startsWith("<") ? "text/xml" : "application/json");
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write(BODY.getBytes(StandardCharsets.UTF_8));
+ }
+ assertTrue(conn.getResponseCode() < 400, "callback must be accepted, got "
+ + conn.getResponseCode());
+ conn.disconnect();
+
+ List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "callback must surface as exactly one event");
+ CloudEvent event = events.get(0);
+ assertEquals("dingtalk.message", event.getType());
+ assertEquals("c-1", event.getSubject());
+ assertTrue(event.getId().startsWith("dingtalk-"),
+ "event id must be namespaced by the platform, got: " + event.getId());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-http/src/test/java/org/apache/eventmesh/connector/http/sink/HttpSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-http/src/test/java/org/apache/eventmesh/connector/http/sink/HttpSinkConnectorTest.java
new file mode 100644
index 0000000..a92b697
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-http/src/test/java/org/apache/eventmesh/connector/http/sink/HttpSinkConnectorTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.http.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {class} POSTs every event's data bytes to the configured URL and swallows
+ * individual failures (log-and-continue contract). A local HTTP server captures the requests.
+ */
+class HttpSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private HttpSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(200, 0);
+ ex.close();
+ });
+ server.start();
+ sink = new HttpSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.url", "http://127.0.0.1:" + server.getAddress().getPort() + "/hook");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putPostsEveryEventData() {
+ sink.put(List.of(event("e1", "alpha"), event("e2", "beta")));
+ assertEquals(List.of("alpha", "beta"), bodies);
+ }
+
+ @Test
+ void putWithEmptyBatchIsNoOp() {
+ sink.put(Collections.emptyList());
+ assertEquals(Collections.emptyList(), bodies);
+ }
+
+ @Test
+ void putWithNullDataPostsEmptyBody() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("n1").withSource(java.net.URI.create("/test"))
+ .withType("test.event").build()));
+ assertEquals(1, bodies.size());
+ assertEquals("", bodies.get(0));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-http/src/test/java/org/apache/eventmesh/connector/http/source/HttpSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-http/src/test/java/org/apache/eventmesh/connector/http/source/HttpSourceConnectorTest.java
new file mode 100644
index 0000000..a46c8cb
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-http/src/test/java/org/apache/eventmesh/connector/http/source/HttpSourceConnectorTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.http.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: HttpSourceConnector receives a POST on its webhook endpoint and poll() surfaces the body
+ * as a CloudEvent (POST -> buffered -> poll contract).
+ */
+class HttpSourceConnectorTest {
+
+ private HttpSourceConnector source;
+ private HttpServer probe; // keeps an unrelated port warm so stop ordering is deterministic
+
+ @BeforeEach
+ void boot() throws Exception {
+ source = new HttpSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/webhook");
+ source.init(props);
+ probe = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ probe.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ probe.stop(0);
+ }
+
+ private int hookPort() throws Exception {
+ // the connector binds its port lazily on first poll; trigger it then discover the bound port
+ java.lang.reflect.Field f = HttpSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ com.sun.net.httpserver.HttpServer s =
+ (com.sun.net.httpserver.HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ @Test
+ void postedBodyIsPolledAsEvent() throws Exception {
+ source.poll(); // trigger lazy server start
+ int port = hookPort();
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/webhook").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write("hello-http".getBytes(StandardCharsets.UTF_8));
+ }
+ assertEquals(200, conn.getResponseCode());
+ conn.disconnect();
+
+ java.util.List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "posted body must surface as exactly one event");
+ assertEquals("hello-http",
+ new String(events.get(0).getData().toBytes(), StandardCharsets.UTF_8));
+ assertTrue(events.get(0).getId() != null && !events.get(0).getId().isEmpty());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/sink/JdbcSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/sink/JdbcSinkConnectorTest.java
new file mode 100644
index 0000000..31e34c0
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/sink/JdbcSinkConnectorTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.jdbc.sink;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for JdbcSinkConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class JdbcSinkConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ JdbcSinkConnector connector = new JdbcSinkConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(Collections.singletonList(last)); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnectorTest.java
new file mode 100644
index 0000000..bf9c90b
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-jdbc/src/test/java/org/apache/eventmesh/connector/jdbc/source/JdbcSourceConnectorTest.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.jdbc.source;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for JdbcSourceConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class JdbcSourceConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ JdbcSourceConnector connector = new JdbcSourceConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(last); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-knative/src/test/java/org/apache/eventmesh/connector/knative/sink/KnativeSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-knative/src/test/java/org/apache/eventmesh/connector/knative/sink/KnativeSinkConnectorTest.java
new file mode 100644
index 0000000..df2cf31
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-knative/src/test/java/org/apache/eventmesh/connector/knative/sink/KnativeSinkConnectorTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.knative.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {class} POSTs every event's data bytes to the configured URL and swallows
+ * individual failures (log-and-continue contract). A local HTTP server captures the requests.
+ */
+class KnativeSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private KnativeSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(200, 0);
+ ex.close();
+ });
+ server.start();
+ sink = new KnativeSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.sinkUrl", "http://127.0.0.1:" + server.getAddress().getPort() + "/hook");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putPostsEveryEventData() {
+ sink.put(List.of(event("e1", "alpha"), event("e2", "beta")));
+ assertEquals(List.of("alpha", "beta"), bodies);
+ }
+
+ @Test
+ void putWithEmptyBatchIsNoOp() {
+ sink.put(Collections.emptyList());
+ assertEquals(Collections.emptyList(), bodies);
+ }
+
+ @Test
+ void putWithNullDataPostsEmptyBody() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("n1").withSource(java.net.URI.create("/test"))
+ .withType("test.event").build()));
+ assertEquals(1, bodies.size());
+ assertEquals("", bodies.get(0));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-knative/src/test/java/org/apache/eventmesh/connector/knative/source/KnativeSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-knative/src/test/java/org/apache/eventmesh/connector/knative/source/KnativeSourceConnectorTest.java
new file mode 100644
index 0000000..5f46bf7
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-knative/src/test/java/org/apache/eventmesh/connector/knative/source/KnativeSourceConnectorTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.knative.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: KnativeSourceConnector receives a POST on its webhook endpoint and poll() surfaces the body
+ * as a CloudEvent (POST -> buffered -> poll contract).
+ */
+class KnativeSourceConnectorTest {
+
+ private KnativeSourceConnector source;
+ private HttpServer probe; // keeps an unrelated port warm so stop ordering is deterministic
+
+ @BeforeEach
+ void boot() throws Exception {
+ source = new KnativeSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/");
+ source.init(props);
+ probe = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ probe.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ probe.stop(0);
+ }
+
+ private int hookPort() throws Exception {
+ // the connector binds its port lazily on first poll; trigger it then discover the bound port
+ java.lang.reflect.Field f = KnativeSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ com.sun.net.httpserver.HttpServer s =
+ (com.sun.net.httpserver.HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ @Test
+ void postedBodyIsPolledAsEvent() throws Exception {
+ source.poll(); // trigger lazy server start
+ int port = hookPort();
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write("hello-knative".getBytes(StandardCharsets.UTF_8));
+ }
+ assertEquals(200, conn.getResponseCode());
+ conn.disconnect();
+
+ java.util.List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "posted body must surface as exactly one event");
+ assertEquals("hello-knative",
+ new String(events.get(0).getData().toBytes(), StandardCharsets.UTF_8));
+ assertTrue(events.get(0).getId() != null && !events.get(0).getId().isEmpty());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-lark/src/test/java/org/apache/eventmesh/connector/lark/sink/LarkSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-lark/src/test/java/org/apache/eventmesh/connector/lark/sink/LarkSinkConnectorTest.java
new file mode 100644
index 0000000..4220be9
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-lark/src/test/java/org/apache/eventmesh/connector/lark/sink/LarkSinkConnectorTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.lark.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {class} POSTs every event's data bytes to the configured URL and swallows
+ * individual failures (log-and-continue contract). A local HTTP server captures the requests.
+ */
+class LarkSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private LarkSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(200, 0);
+ ex.close();
+ });
+ server.start();
+ sink = new LarkSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.webhookUrl", "http://127.0.0.1:" + server.getAddress().getPort() + "/hook");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putPostsEveryEventData() {
+ sink.put(List.of(event("e1", "alpha"), event("e2", "beta")));
+ assertEquals(List.of("alpha", "beta"), bodies);
+ }
+
+ @Test
+ void putWithEmptyBatchIsNoOp() {
+ sink.put(Collections.emptyList());
+ assertEquals(Collections.emptyList(), bodies);
+ }
+
+ @Test
+ void putWithNullDataPostsEmptyBody() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("n1").withSource(java.net.URI.create("/test"))
+ .withType("test.event").build()));
+ assertEquals(1, bodies.size());
+ assertEquals("", bodies.get(0));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-lark/src/test/java/org/apache/eventmesh/connector/lark/source/LarkSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-lark/src/test/java/org/apache/eventmesh/connector/lark/source/LarkSourceConnectorTest.java
new file mode 100644
index 0000000..107f014
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-lark/src/test/java/org/apache/eventmesh/connector/lark/source/LarkSourceConnectorTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.lark.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: LarkSourceConnector lazily boots its webhook endpoint on first poll; a native lark
+ * callback payload POSTed to the hook surfaces via poll() as a CloudEvent carrying the platform
+ * message id / subject mapping.
+ */
+class LarkSourceConnectorTest {
+
+ private static final String BODY = "{\"header\":{\"event_id\":\"ev-1\",\"event_type\":\"im.message.receive_v1\",\"app_id\":\"app-1\"}}";
+
+ private int hookPort(LarkSourceConnector source) throws Exception {
+ Field f = LarkSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ HttpServer s = (HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ private LarkSourceConnector boot() {
+ LarkSourceConnector source = new LarkSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/lark");
+ // no secret/token set -> signature checks are disabled (dev mode)
+ source.init(props);
+ return source;
+ }
+
+ @Test
+ void postedCallbackSurfacesAsEvent() throws Exception {
+ LarkSourceConnector source = boot();
+ source.poll(); // trigger lazy server start
+ int port = hookPort(source);
+
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/lark").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setRequestProperty("Content-Type",
+ BODY.startsWith("<") ? "text/xml" : "application/json");
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write(BODY.getBytes(StandardCharsets.UTF_8));
+ }
+ assertTrue(conn.getResponseCode() < 400, "callback must be accepted, got "
+ + conn.getResponseCode());
+ conn.disconnect();
+
+ List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "callback must surface as exactly one event");
+ CloudEvent event = events.get(0);
+ assertEquals("lark.im.message.receive_v1", event.getType());
+ assertEquals("app-1", event.getSubject());
+ assertTrue(event.getId().startsWith("lark-"),
+ "event id must be namespaced by the platform, got: " + event.getId());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mcp/src/test/java/org/apache/eventmesh/connector/mcp/sink/McpSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-mcp/src/test/java/org/apache/eventmesh/connector/mcp/sink/McpSinkConnectorTest.java
new file mode 100644
index 0000000..80611b1
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-mcp/src/test/java/org/apache/eventmesh/connector/mcp/sink/McpSinkConnectorTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.mcp.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {@link McpSinkConnector} wraps each event's data into a JSON-RPC 2.0
+ * request (method from config, monotonically increasing id) and POSTs it to the MCP server URL;
+ * non-2xx answers throw (redelivery contract).
+ */
+class McpSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private final AtomicInteger status = new AtomicInteger(200);
+ private McpSinkConnector sink;
+ private final ObjectMapper mapper = new ObjectMapper();
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(status.get(), 0);
+ ex.close();
+ });
+ server.start();
+ sink = new McpSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.mcpServerUrl",
+ "http://127.0.0.1:" + server.getAddress().getPort() + "/mcp");
+ props.setProperty("connector.method", "eventmesh.notify");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putWrapsDataInJsonRpcEnvelope() throws Exception {
+ sink.put(Collections.singletonList(event("e1", "{\"k\":\"v\"}")));
+ assertEquals(1, bodies.size());
+ JsonNode rpc = mapper.readTree(bodies.get(0));
+ assertEquals("2.0", rpc.get("jsonrpc").asText());
+ assertEquals("eventmesh.notify", rpc.get("method").asText());
+ assertEquals("v", rpc.get("params").get("k").asText());
+ assertTrue(rpc.get("id").asInt() >= 1, "request id must be assigned");
+ }
+
+ @Test
+ void putThrowsOnNon2xx() {
+ status.set(500);
+ assertThrows(RuntimeException.class,
+ () -> sink.put(Collections.singletonList(event("e2", "{}"))));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mcp/src/test/java/org/apache/eventmesh/connector/mcp/source/McpSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-mcp/src/test/java/org/apache/eventmesh/connector/mcp/source/McpSourceConnectorTest.java
new file mode 100644
index 0000000..d4eb395
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-mcp/src/test/java/org/apache/eventmesh/connector/mcp/source/McpSourceConnectorTest.java
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.mcp.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: McpSourceConnector receives a POST on its webhook endpoint and poll() surfaces the body
+ * as a CloudEvent (POST -> buffered -> poll contract).
+ */
+class McpSourceConnectorTest {
+
+ private McpSourceConnector source;
+ private HttpServer probe; // keeps an unrelated port warm so stop ordering is deterministic
+
+ @BeforeEach
+ void boot() throws Exception {
+ source = new McpSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/mcp");
+ source.init(props);
+ probe = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ probe.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ probe.stop(0);
+ }
+
+ private int hookPort() throws Exception {
+ // the connector binds its port lazily on first poll; trigger it then discover the bound port
+ java.lang.reflect.Field f = McpSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ com.sun.net.httpserver.HttpServer s =
+ (com.sun.net.httpserver.HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ @Test
+ void postedBodyIsPolledAsEvent() throws Exception {
+ source.poll(); // trigger lazy server start
+ int port = hookPort();
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/mcp").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write("{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"notify\"}".getBytes(StandardCharsets.UTF_8));
+ }
+ org.junit.jupiter.api.Assertions.assertTrue(conn.getResponseCode() < 300,
+ "callback must be accepted (2xx), got " + conn.getResponseCode());
+ conn.disconnect();
+
+ java.util.List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "posted body must surface as exactly one event");
+ assertEquals("{\"jsonrpc\":\"2.0\",\"id\":1,\"method\":\"notify\"}",
+ new String(events.get(0).getData().toBytes(), StandardCharsets.UTF_8));
+ assertTrue(events.get(0).getId() != null && !events.get(0).getId().isEmpty());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mongodb/src/test/java/org/apache/eventmesh/connector/mongodb/sink/MongodbSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-mongodb/src/test/java/org/apache/eventmesh/connector/mongodb/sink/MongodbSinkConnectorTest.java
new file mode 100644
index 0000000..17b9493
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-mongodb/src/test/java/org/apache/eventmesh/connector/mongodb/sink/MongodbSinkConnectorTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.mongodb.sink;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for MongodbSinkConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class MongodbSinkConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ MongodbSinkConnector connector = new MongodbSinkConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(Collections.singletonList(last)); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-mongodb/src/test/java/org/apache/eventmesh/connector/mongodb/source/MongodbSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-mongodb/src/test/java/org/apache/eventmesh/connector/mongodb/source/MongodbSourceConnectorTest.java
new file mode 100644
index 0000000..7a51cb2
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-mongodb/src/test/java/org/apache/eventmesh/connector/mongodb/source/MongodbSourceConnectorTest.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.mongodb.source;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for MongodbSourceConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class MongodbSourceConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ MongodbSourceConnector connector = new MongodbSourceConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(last); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/test/java/org/apache/eventmesh/connector/openfunction/sink/OpenfunctionSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/test/java/org/apache/eventmesh/connector/openfunction/sink/OpenfunctionSinkConnectorTest.java
new file mode 100644
index 0000000..cc05539
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/test/java/org/apache/eventmesh/connector/openfunction/sink/OpenfunctionSinkConnectorTest.java
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.openfunction.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {@link OpenfunctionSinkConnector} POSTs each event to the function URL with
+ * the CloudEvents context in Ce-Id/Ce-Type/Ce-Source headers; non-2xx throws (redelivery).
+ */
+class OpenfunctionSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private final List<String> ceIds = new CopyOnWriteArrayList<>();
+ private final AtomicInteger status = new AtomicInteger(200);
+ private OpenfunctionSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ceIds.add(ex.getRequestHeaders().getFirst("Ce-Id"));
+ ex.sendResponseHeaders(status.get(), 0);
+ ex.close();
+ });
+ server.start();
+ sink = new OpenfunctionSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.functionUrl",
+ "http://127.0.0.1:" + server.getAddress().getPort() + "/fn");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putForwardsEventWithCeContextHeaders() {
+ sink.put(Collections.singletonList(event("of-1", "payload")));
+ assertEquals(List.of("payload"), bodies);
+ assertEquals(List.of("of-1"), ceIds, "Ce-Id header must carry the event id");
+ }
+
+ @Test
+ void putThrowsOnNon2xx() {
+ status.set(502);
+ assertThrows(RuntimeException.class,
+ () -> sink.put(Collections.singletonList(event("of-2", "x"))));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/test/java/org/apache/eventmesh/connector/openfunction/source/OpenfunctionSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/test/java/org/apache/eventmesh/connector/openfunction/source/OpenfunctionSourceConnectorTest.java
new file mode 100644
index 0000000..5a6929f
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-openfunction/src/test/java/org/apache/eventmesh/connector/openfunction/source/OpenfunctionSourceConnectorTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.openfunction.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.InetSocketAddress;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: OpenfunctionSourceConnector receives a POST on its webhook endpoint and poll() surfaces the body
+ * as a CloudEvent (POST -> buffered -> poll contract).
+ */
+class OpenfunctionSourceConnectorTest {
+
+ private OpenfunctionSourceConnector source;
+ private HttpServer probe; // keeps an unrelated port warm so stop ordering is deterministic
+
+ @BeforeEach
+ void boot() throws Exception {
+ source = new OpenfunctionSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/openfunction");
+ source.init(props);
+ probe = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ probe.start();
+ }
+
+ @AfterEach
+ void tearDown() {
+ probe.stop(0);
+ }
+
+ private int hookPort() throws Exception {
+ // the connector binds its port lazily on first poll; trigger it then discover the bound port
+ java.lang.reflect.Field f = OpenfunctionSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ com.sun.net.httpserver.HttpServer s =
+ (com.sun.net.httpserver.HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ @Test
+ void postedBodyIsPolledAsEvent() throws Exception {
+ source.poll(); // trigger lazy server start
+ int port = hookPort();
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/openfunction").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write("{\"x\":1}".getBytes(StandardCharsets.UTF_8));
+ }
+ assertEquals(200, conn.getResponseCode());
+ conn.disconnect();
+
+ java.util.List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "posted body must surface as exactly one event");
+ assertEquals("{\"x\":1}",
+ new String(events.get(0).getData().toBytes(), StandardCharsets.UTF_8));
+ assertTrue(events.get(0).getId() != null && !events.get(0).getId().isEmpty());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/test/java/org/apache/eventmesh/connector/pravega/sink/PravegaSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/test/java/org/apache/eventmesh/connector/pravega/sink/PravegaSinkConnectorTest.java
new file mode 100644
index 0000000..8fe6e91
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/test/java/org/apache/eventmesh/connector/pravega/sink/PravegaSinkConnectorTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.pravega.sink;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for PravegaSinkConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class PravegaSinkConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ PravegaSinkConnector connector = new PravegaSinkConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(Collections.singletonList(last)); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pravega/src/test/java/org/apache/eventmesh/connector/pravega/source/PravegaSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/test/java/org/apache/eventmesh/connector/pravega/source/PravegaSourceConnectorTest.java
new file mode 100644
index 0000000..89d4ae6
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-pravega/src/test/java/org/apache/eventmesh/connector/pravega/source/PravegaSourceConnectorTest.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.pravega.source;
+
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for PravegaSourceConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class PravegaSourceConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ PravegaSourceConnector connector = new PravegaSourceConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(last); // must not throw
+ }
+
+ @Test
+ void initParsesDocumentedDefaults() {
+ PravegaSourceConnector connector = new PravegaSourceConnector();
+ connector.init(new Properties()); // must not connect anything
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/test/java/org/apache/eventmesh/connector/prometheus/sink/PrometheusSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/test/java/org/apache/eventmesh/connector/prometheus/sink/PrometheusSinkConnectorTest.java
new file mode 100644
index 0000000..4a76efe
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/test/java/org/apache/eventmesh/connector/prometheus/sink/PrometheusSinkConnectorTest.java
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.prometheus.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {@link PrometheusSinkConnector} merges the batch into ONE text exposition
+ * payload pushed to the Pushgateway in a single request; an all-empty batch skips the push; a
+ * non-2xx answer throws (redelivery contract).
+ */
+class PrometheusSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private final AtomicInteger status = new AtomicInteger(200);
+ private PrometheusSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(status.get(), 0);
+ ex.close();
+ });
+ server.start();
+ sink = new PrometheusSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.pushgatewayUrl",
+ "http://127.0.0.1:" + server.getAddress().getPort());
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent metric(String data) {
+ return CloudEventBuilder.v1().withId("m").withSource(java.net.URI.create("/test"))
+ .withType("test.metric")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putMergesBatchIntoOnePush() {
+ sink.put(Arrays.asList(metric("a 1\n"), metric("b 2\n")));
+ assertEquals(1, bodies.size(), "batch must be one merged push");
+ String payload = bodies.get(0);
+ org.junit.jupiter.api.Assertions.assertTrue(payload.contains("a 1"));
+ org.junit.jupiter.api.Assertions.assertTrue(payload.contains("b 2"));
+ }
+
+ @Test
+ void putWithAllEmptyDataSkipsPush() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("m").withSource(java.net.URI.create("/t"))
+ .withType("test.metric").build()));
+ assertEquals(Collections.emptyList(), bodies, "no metric text -> no push");
+ }
+
+ @Test
+ void putThrowsOnNon2xx() {
+ status.set(500);
+ assertThrows(RuntimeException.class, () -> sink.put(List.of(metric("x 1\n"))));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/test/java/org/apache/eventmesh/connector/prometheus/source/PrometheusSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/test/java/org/apache/eventmesh/connector/prometheus/source/PrometheusSourceConnectorTest.java
new file mode 100644
index 0000000..369ecea
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-prometheus/src/test/java/org/apache/eventmesh/connector/prometheus/source/PrometheusSourceConnectorTest.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.prometheus.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.OutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {@link PrometheusSourceConnector} scrapes the configured metrics URL and
+ * surfaces the exposition body as one CloudEvent per poll.
+ */
+class PrometheusSourceConnectorTest {
+
+ private HttpServer server;
+ private PrometheusSourceConnector source;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/metrics", (HttpExchange ex) -> {
+ byte[] body = "up 1\n".getBytes(StandardCharsets.UTF_8);
+ ex.sendResponseHeaders(200, body.length);
+ try (OutputStream os = ex.getResponseBody()) {
+ os.write(body);
+ }
+ });
+ server.start();
+ source = new PrometheusSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.metricsUrl",
+ "http://127.0.0.1:" + server.getAddress().getPort() + "/metrics");
+ source.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ server.stop(0);
+ }
+
+ @Test
+ void pollScrapesMetricsBodyIntoEvent() {
+ List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size());
+ assertEquals("up 1\n",
+ new String(events.get(0).getData().toBytes(), StandardCharsets.UTF_8));
+ assertEquals("prometheus.metrics", events.get(0).getType());
+ }
+
+ @Test
+ void pollOnUnreachableUrlReturnsEmpty() {
+ PrometheusSourceConnector dead = new PrometheusSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.metricsUrl", "http://127.0.0.1:1/metrics");
+ dead.init(props);
+ org.junit.jupiter.api.Assertions.assertTrue(dead.poll().isEmpty(),
+ "scrape failure must degrade to an empty poll, not throw");
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/src/test/java/org/apache/eventmesh/connector/rabbitmq/sink/RabbitmqSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/src/test/java/org/apache/eventmesh/connector/rabbitmq/sink/RabbitmqSinkConnectorTest.java
new file mode 100644
index 0000000..7ecf7c1
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/src/test/java/org/apache/eventmesh/connector/rabbitmq/sink/RabbitmqSinkConnectorTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.rabbitmq.sink;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for RabbitmqSinkConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class RabbitmqSinkConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ RabbitmqSinkConnector connector = new RabbitmqSinkConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(Collections.singletonList(last)); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/src/test/java/org/apache/eventmesh/connector/rabbitmq/source/RabbitmqSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/src/test/java/org/apache/eventmesh/connector/rabbitmq/source/RabbitmqSourceConnectorTest.java
new file mode 100644
index 0000000..bc984bd
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-rabbitmq/src/test/java/org/apache/eventmesh/connector/rabbitmq/source/RabbitmqSourceConnectorTest.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.rabbitmq.source;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for RabbitmqSourceConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class RabbitmqSourceConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ RabbitmqSourceConnector connector = new RabbitmqSourceConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(last); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/RedisSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/RedisSinkConnectorTest.java
new file mode 100644
index 0000000..0f7dee0
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/sink/RedisSinkConnectorTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.redis.sink;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for RedisSinkConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class RedisSinkConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ RedisSinkConnector connector = new RedisSinkConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(Collections.singletonList(last)); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/source/RedisSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/source/RedisSourceConnectorTest.java
new file mode 100644
index 0000000..d8f8df0
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-redis/src/test/java/org/apache/eventmesh/connector/redis/source/RedisSourceConnectorTest.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.redis.source;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for RedisSourceConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class RedisSourceConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ RedisSourceConnector connector = new RedisSourceConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(last); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/sink/S3SinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/sink/S3SinkConnectorTest.java
new file mode 100644
index 0000000..60ad9ae
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/sink/S3SinkConnectorTest.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.s3.sink;
+
+import java.util.Collections;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for S3SinkConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class S3SinkConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ S3SinkConnector connector = new S3SinkConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(Collections.singletonList(last)); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/source/S3SourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/source/S3SourceConnectorTest.java
new file mode 100644
index 0000000..c5140c3
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-s3/src/test/java/org/apache/eventmesh/connector/s3/source/S3SourceConnectorTest.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.s3.source;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Contract-level unit test for S3SourceConnector: the external system is NOT available in CI, so this covers
+ * the parts that must hold regardless — instantiation, the documented default-config surface of
+ * init() (where init only parses config), and the commit() no-op contract.
+ */
+class S3SourceConnectorTest {
+
+ @Test
+ void commitIsNoOp() {
+ S3SourceConnector connector = new S3SourceConnector();
+ CloudEvent last = CloudEventBuilder.v1().withId("e1")
+ .withSource(java.net.URI.create("/test")).withType("test.event").build();
+ connector.commit(last); // must not throw
+ }
+
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-slack/src/test/java/org/apache/eventmesh/connector/slack/sink/SlackSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-slack/src/test/java/org/apache/eventmesh/connector/slack/sink/SlackSinkConnectorTest.java
new file mode 100644
index 0000000..28691e9
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-slack/src/test/java/org/apache/eventmesh/connector/slack/sink/SlackSinkConnectorTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.slack.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {class} POSTs every event's data bytes to the configured URL and swallows
+ * individual failures (log-and-continue contract). A local HTTP server captures the requests.
+ */
+class SlackSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private SlackSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(200, 0);
+ ex.close();
+ });
+ server.start();
+ sink = new SlackSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.webhookUrl", "http://127.0.0.1:" + server.getAddress().getPort() + "/hook");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putPostsEveryEventData() {
+ sink.put(List.of(event("e1", "alpha"), event("e2", "beta")));
+ assertEquals(List.of("alpha", "beta"), bodies);
+ }
+
+ @Test
+ void putWithEmptyBatchIsNoOp() {
+ sink.put(Collections.emptyList());
+ assertEquals(Collections.emptyList(), bodies);
+ }
+
+ @Test
+ void putWithNullDataPostsEmptyBody() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("n1").withSource(java.net.URI.create("/test"))
+ .withType("test.event").build()));
+ assertEquals(1, bodies.size());
+ assertEquals("", bodies.get(0));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-slack/src/test/java/org/apache/eventmesh/connector/slack/source/SlackSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-slack/src/test/java/org/apache/eventmesh/connector/slack/source/SlackSourceConnectorTest.java
new file mode 100644
index 0000000..0f056e2
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-slack/src/test/java/org/apache/eventmesh/connector/slack/source/SlackSourceConnectorTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.slack.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: SlackSourceConnector lazily boots its webhook endpoint on first poll; a native slack
+ * callback payload POSTed to the hook surfaces via poll() as a CloudEvent carrying the platform
+ * message id / subject mapping.
+ */
+class SlackSourceConnectorTest {
+
+ private static final String BODY = "{\"event_id\":\"ev-9\",\"team_id\":\"T-1\",\"event\":{\"type\":\"message\"}}";
+
+ private int hookPort(SlackSourceConnector source) throws Exception {
+ Field f = SlackSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ HttpServer s = (HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ private SlackSourceConnector boot() {
+ SlackSourceConnector source = new SlackSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/slack");
+ // no secret/token set -> signature checks are disabled (dev mode)
+ source.init(props);
+ return source;
+ }
+
+ @Test
+ void postedCallbackSurfacesAsEvent() throws Exception {
+ SlackSourceConnector source = boot();
+ source.poll(); // trigger lazy server start
+ int port = hookPort(source);
+
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/slack").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setRequestProperty("Content-Type",
+ BODY.startsWith("<") ? "text/xml" : "application/json");
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write(BODY.getBytes(StandardCharsets.UTF_8));
+ }
+ assertTrue(conn.getResponseCode() < 400, "callback must be accepted, got "
+ + conn.getResponseCode());
+ conn.disconnect();
+
+ List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "callback must surface as exactly one event");
+ CloudEvent event = events.get(0);
+ assertEquals("slack.message", event.getType());
+ assertEquals("T-1", event.getSubject());
+ assertTrue(event.getId().startsWith("slack-"),
+ "event id must be namespaced by the platform, got: " + event.getId());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-spring/src/test/java/org/apache/eventmesh/connector/spring/sink/SpringSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-spring/src/test/java/org/apache/eventmesh/connector/spring/sink/SpringSinkConnectorTest.java
new file mode 100644
index 0000000..0947281
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-spring/src/test/java/org/apache/eventmesh/connector/spring/sink/SpringSinkConnectorTest.java
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.spring.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Hermetic unit test of the {@link SpringSinkConnector} forwarding contract: put() without an
+ * injected EventForwarder fails fast; with one, every event is forwarded in order and a failing
+ * forwarder surfaces as a RuntimeException (redelivery).
+ */
+class SpringSinkConnectorTest {
+
+ private static CloudEvent event(String id) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event").build();
+ }
+
+ @Test
+ void putWithoutForwarderFailsFast() {
+ SpringSinkConnector sink = new SpringSinkConnector();
+ sink.init(new Properties());
+ assertThrows(IllegalStateException.class,
+ () -> sink.put(Collections.singletonList(event("e1"))));
+ }
+
+ @Test
+ void putForwardsEveryEventInOrder() {
+ SpringSinkConnector sink = new SpringSinkConnector();
+ sink.init(new Properties());
+ List<CloudEvent> seen = new CopyOnWriteArrayList<>();
+ sink.setForwarder(seen::add);
+ sink.put(Arrays.asList(event("e1"), event("e2"), event("e3")));
+ assertEquals(Arrays.asList("e1", "e2", "e3"),
+ seen.stream().map(CloudEvent::getId).toList());
+ }
+
+ @Test
+ void putSurfacesForwarderFailure() {
+ SpringSinkConnector sink = new SpringSinkConnector();
+ sink.init(new Properties());
+ sink.setForwarder(e -> {
+ throw new IllegalArgumentException("boom");
+ });
+ RuntimeException ex = assertThrows(RuntimeException.class,
+ () -> sink.put(Collections.singletonList(event("e1"))));
+ assertTrue(ex.getMessage().contains("e1"), "error must name the failing event");
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-spring/src/test/java/org/apache/eventmesh/connector/spring/source/SpringSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-spring/src/test/java/org/apache/eventmesh/connector/spring/source/SpringSourceConnectorTest.java
new file mode 100644
index 0000000..e6dc1c5
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-spring/src/test/java/org/apache/eventmesh/connector/spring/source/SpringSourceConnectorTest.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.spring.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.lang.reflect.Field;
+import java.net.URI;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.LinkedBlockingQueue;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Hermetic unit test of the {@link SpringSourceConnector} buffer contract: a Spring
+ * {@code @EventListener} feeds the internal buffer (simulated here directly); poll() drains it
+ * in order and is a no-op before init.
+ */
+class SpringSourceConnectorTest {
+
+ private static CloudEvent event(String id) {
+ return CloudEventBuilder.v1().withId(id).withSource(URI.create("/test"))
+ .withType("test.event").build();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void pollDrainsBufferedInOrder() throws Exception {
+ SpringSourceConnector source = new SpringSourceConnector();
+ source.init(new Properties());
+
+ Field f = SpringSourceConnector.class.getDeclaredField("buffer");
+ f.setAccessible(true);
+ LinkedBlockingQueue<CloudEvent> buffer = (LinkedBlockingQueue<CloudEvent>) f.get(source);
+ buffer.offer(event("e1"));
+ buffer.offer(event("e2"));
+
+ List<CloudEvent> drained = source.poll();
+ assertEquals(List.of("e1", "e2"), drained.stream().map(CloudEvent::getId).toList());
+ assertEquals(0, source.poll().size(), "second poll drains nothing");
+ }
+
+ @Test
+ void pollBeforeInitReturnsEmpty() {
+ SpringSourceConnector fresh = new SpringSourceConnector();
+ assertTrue(fresh.poll().isEmpty());
+ }
+
+ @Test
+ void commitIsNoOp() {
+ SpringSourceConnector source = new SpringSourceConnector();
+ source.init(new Properties());
+ source.commit(event("e1")); // must not throw
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wechat/src/test/java/org/apache/eventmesh/connector/wechat/sink/WechatSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-wechat/src/test/java/org/apache/eventmesh/connector/wechat/sink/WechatSinkConnectorTest.java
new file mode 100644
index 0000000..d4b25b6
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-wechat/src/test/java/org/apache/eventmesh/connector/wechat/sink/WechatSinkConnectorTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.wechat.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {class} POSTs every event's data bytes to the configured URL and swallows
+ * individual failures (log-and-continue contract). A local HTTP server captures the requests.
+ */
+class WechatSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private WechatSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(200, 0);
+ ex.close();
+ });
+ server.start();
+ sink = new WechatSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.webhookUrl", "http://127.0.0.1:" + server.getAddress().getPort() + "/hook");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putPostsEveryEventData() {
+ sink.put(List.of(event("e1", "alpha"), event("e2", "beta")));
+ assertEquals(List.of("alpha", "beta"), bodies);
+ }
+
+ @Test
+ void putWithEmptyBatchIsNoOp() {
+ sink.put(Collections.emptyList());
+ assertEquals(Collections.emptyList(), bodies);
+ }
+
+ @Test
+ void putWithNullDataPostsEmptyBody() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("n1").withSource(java.net.URI.create("/test"))
+ .withType("test.event").build()));
+ assertEquals(1, bodies.size());
+ assertEquals("", bodies.get(0));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wechat/src/test/java/org/apache/eventmesh/connector/wechat/source/WechatSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-wechat/src/test/java/org/apache/eventmesh/connector/wechat/source/WechatSourceConnectorTest.java
new file mode 100644
index 0000000..0d074ea
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-wechat/src/test/java/org/apache/eventmesh/connector/wechat/source/WechatSourceConnectorTest.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.wechat.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: WechatSourceConnector lazily boots its webhook endpoint on first poll; a native wechat
+ * callback payload POSTed to the hook surfaces via poll() as a CloudEvent carrying the platform
+ * message id / subject mapping.
+ */
+class WechatSourceConnectorTest {
+
+ private static final String BODY = "<xml><ToUserName><![CDATA[gh_1]]></ToUserName>"
+ + "<FromUserName><![CDATA[o-1]]></FromUserName>"
+ + "<MsgType><![CDATA[text]]></MsgType>"
+ + "<Content><![CDATA[hello-wechat]]></Content>"
+ + "<MsgId>123</MsgId></xml>";
+
+ private int hookPort(WechatSourceConnector source) throws Exception {
+ Field f = WechatSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ HttpServer s = (HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ private WechatSourceConnector boot() {
+ WechatSourceConnector source = new WechatSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/wechat");
+ // no secret/token set -> signature checks are disabled (dev mode)
+ source.init(props);
+ return source;
+ }
+
+ @Test
+ void postedCallbackSurfacesAsEvent() throws Exception {
+ WechatSourceConnector source = boot();
+ source.poll(); // trigger lazy server start
+ int port = hookPort(source);
+
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/wechat").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setRequestProperty("Content-Type",
+ BODY.startsWith("<") ? "text/xml" : "application/json");
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write(BODY.getBytes(StandardCharsets.UTF_8));
+ }
+ assertTrue(conn.getResponseCode() < 400, "callback must be accepted, got "
+ + conn.getResponseCode());
+ conn.disconnect();
+
+ List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "callback must surface as exactly one event");
+ CloudEvent event = events.get(0);
+ assertEquals("wechat.text", event.getType());
+ assertEquals("o-1", event.getSubject());
+ assertTrue(event.getId().startsWith("wechat-"),
+ "event id must be namespaced by the platform, got: " + event.getId());
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wecom/src/test/java/org/apache/eventmesh/connector/wecom/sink/WecomSinkConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-wecom/src/test/java/org/apache/eventmesh/connector/wecom/sink/WecomSinkConnectorTest.java
new file mode 100644
index 0000000..e9753cf
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-wecom/src/test/java/org/apache/eventmesh/connector/wecom/sink/WecomSinkConnectorTest.java
@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.wecom.sink;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.ByteArrayOutputStream;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: {class} POSTs every event's data bytes to the configured URL and swallows
+ * individual failures (log-and-continue contract). A local HTTP server captures the requests.
+ */
+class WecomSinkConnectorTest {
+
+ private HttpServer server;
+ private final List<String> bodies = new CopyOnWriteArrayList<>();
+ private WecomSinkConnector sink;
+
+ @BeforeEach
+ void boot() throws Exception {
+ server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+ server.createContext("/", (HttpExchange ex) -> {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ ex.getRequestBody().transferTo(out);
+ bodies.add(out.toString(StandardCharsets.UTF_8));
+ ex.sendResponseHeaders(200, 0);
+ ex.close();
+ });
+ server.start();
+ sink = new WecomSinkConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.webhookUrl", "http://127.0.0.1:" + server.getAddress().getPort() + "/hook");
+ sink.init(props);
+ }
+
+ @AfterEach
+ void tearDown() {
+ sink.commit(Collections.emptyList());
+ server.stop(0);
+ }
+
+ private static CloudEvent event(String id, String data) {
+ return CloudEventBuilder.v1().withId(id).withSource(java.net.URI.create("/test"))
+ .withType("test.event")
+ .withData(data.getBytes(StandardCharsets.UTF_8)).build();
+ }
+
+ @Test
+ void putPostsEveryEventData() {
+ sink.put(List.of(event("e1", "alpha"), event("e2", "beta")));
+ assertEquals(List.of("alpha", "beta"), bodies);
+ }
+
+ @Test
+ void putWithEmptyBatchIsNoOp() {
+ sink.put(Collections.emptyList());
+ assertEquals(Collections.emptyList(), bodies);
+ }
+
+ @Test
+ void putWithNullDataPostsEmptyBody() {
+ sink.put(Collections.singletonList(
+ CloudEventBuilder.v1().withId("n1").withSource(java.net.URI.create("/test"))
+ .withType("test.event").build()));
+ assertEquals(1, bodies.size());
+ assertEquals("", bodies.get(0));
+ }
+}
diff --git a/eventmesh-connector-plugin/eventmesh-connector-wecom/src/test/java/org/apache/eventmesh/connector/wecom/source/WecomSourceConnectorTest.java b/eventmesh-connector-plugin/eventmesh-connector-wecom/src/test/java/org/apache/eventmesh/connector/wecom/source/WecomSourceConnectorTest.java
new file mode 100644
index 0000000..a2f47a0
--- /dev/null
+++ b/eventmesh-connector-plugin/eventmesh-connector-wecom/src/test/java/org/apache/eventmesh/connector/wecom/source/WecomSourceConnectorTest.java
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.connector.wecom.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Properties;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+
+import com.sun.net.httpserver.HttpServer;
+
+/**
+ * Hermetic unit test: WecomSourceConnector lazily boots its webhook endpoint on first poll; a native wecom
+ * callback payload POSTed to the hook surfaces via poll() as a CloudEvent carrying the platform
+ * message id / subject mapping.
+ */
+class WecomSourceConnectorTest {
+
+ private static final String BODY = "{\"msgid\":\"w-1\",\"event_type\":\"message\",\"from\":{\"user_id\":\"u-1\"}}";
+
+ private int hookPort(WecomSourceConnector source) throws Exception {
+ Field f = WecomSourceConnector.class.getDeclaredField("server");
+ f.setAccessible(true);
+ HttpServer s = (HttpServer) f.get(source);
+ return s.getAddress().getPort();
+ }
+
+ private WecomSourceConnector boot() {
+ WecomSourceConnector source = new WecomSourceConnector();
+ Properties props = new Properties();
+ props.setProperty("connector.port", "0");
+ props.setProperty("connector.path", "/wecom");
+ // no secret/token set -> signature checks are disabled (dev mode)
+ source.init(props);
+ return source;
+ }
+
+ @Test
+ void postedCallbackSurfacesAsEvent() throws Exception {
+ WecomSourceConnector source = boot();
+ source.poll(); // trigger lazy server start
+ int port = hookPort(source);
+
+ HttpURLConnection conn = (HttpURLConnection) new URL(
+ "http://127.0.0.1:" + port + "/wecom").openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setRequestProperty("Content-Type",
+ BODY.startsWith("<") ? "text/xml" : "application/json");
+ try (OutputStream os = conn.getOutputStream()) {
+ os.write(BODY.getBytes(StandardCharsets.UTF_8));
+ }
+ assertTrue(conn.getResponseCode() < 400, "callback must be accepted, got "
+ + conn.getResponseCode());
+ conn.disconnect();
+
+ List<CloudEvent> events = source.poll();
+ assertEquals(1, events.size(), "callback must surface as exactly one event");
+ CloudEvent event = events.get(0);
+ assertEquals("wecom.message", event.getType());
+ assertEquals("u-1", event.getSubject());
+ assertTrue(event.getId().startsWith("wecom-"),
+ "event id must be namespaced by the platform, got: " + event.getId());
+ }
+}
diff --git a/eventmesh-runtime/build.gradle b/eventmesh-runtime/build.gradle
index 3e01b5e..bc8c417 100644
--- a/eventmesh-runtime/build.gradle
+++ b/eventmesh-runtime/build.gradle
@@ -28,6 +28,12 @@
implementation 'io.cloudevents:cloudevents-core'
implementation 'io.cloudevents:cloudevents-json-jackson'
+ // #5411: legacy SDK gRPC bridge — gRPC runtime (netty-shaded server + protobuf plumbing).
+ // Version aligns with eventmesh-common's grpcVersion (1.68.0).
+ implementation 'io.grpc:grpc-netty-shaded:1.68.0'
+ implementation 'io.grpc:grpc-protobuf:1.68.0'
+ implementation 'io.grpc:grpc-stub:1.68.0'
+
implementation 'io.opentelemetry:opentelemetry-api'
implementation 'io.netty:netty-all'
diff --git a/eventmesh-runtime/conf/eventmesh.properties b/eventmesh-runtime/conf/eventmesh.properties
index a864ebd..85c96f1 100644
--- a/eventmesh-runtime/conf/eventmesh.properties
+++ b/eventmesh-runtime/conf/eventmesh.properties
@@ -52,6 +52,10 @@
#eventmesh.http.port=10105
#eventmesh.admin.port=10106
#eventmesh.ws.port=-1
+# Legacy SDK gRPC bridge (issue #5411): serves PublisherService/ConsumerService/
+# HeartbeatService for old EventMeshGrpcProducer/EventMeshGrpcConsumer clients.
+# Opt-in: unset/-1 = disabled; set 10205 (the 1.x default) to enable.
+#eventmesh.grpc.port=-1
#eventmesh.offset.path=./data/offset
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
index 781d9b5..9f88b73 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/EventMeshApplication.java
@@ -70,6 +70,9 @@
private org.apache.eventmesh.runtime.a2a.A2AGatewayServer a2aGateway;
private org.apache.eventmesh.runtime.a2a.EventMeshA2ATransport a2aTransport;
private org.apache.eventmesh.runtime.state.TaskStore a2aTaskStore;
+ /** #5411: optional legacy SDK gRPC bridge (Publisher/Consumer/Heartbeat on port 10205). */
+ private org.apache.eventmesh.runtime.grpc.EventMeshGrpcServer grpcBridge;
+ private int grpcPort = -1;
private org.apache.eventmesh.runtime.session.AgentRegistrar agentRegistrar;
private org.apache.eventmesh.runtime.session.Matchmaker matchmaker;
private org.apache.eventmesh.runtime.session.SessionRouter sessionRouter;
@@ -187,6 +190,21 @@
return this;
}
+ /**
+ * #5411: enable the legacy SDK gRPC compatibility bridge (PublisherService /
+ * ConsumerService / HeartbeatService on {@code port}, 0 = auto-select). Must be called
+ * before {@link #start()}; the bridge shuts down with the application.
+ */
+ public EventMeshApplication withGrpcBridge(int port) {
+ this.grpcPort = port;
+ return this;
+ }
+
+ /** #5411: the live gRPC bridge (null before {@link #start()} or when not enabled). */
+ public org.apache.eventmesh.runtime.grpc.EventMeshGrpcServer grpcBridge() {
+ return grpcBridge;
+ }
+
public EventMeshApplication(MeshStoragePlugin storage, OffsetStore offsetStore, int httpPort, int adminPort) {
// #5338: read delivery topology from system property (with documented default).
// Missing / blank -> LOCAL_STICKY_PULL (backward compatible). Unknown value -> fail-fast
@@ -365,8 +383,35 @@
log.info("A2A gateway started: port={} taskStore={} auth={}",
a2aGateway.getPort(), a2aTaskStore.getClass().getSimpleName(), "enabled");
}
- log.info("EventMeshApplication started: traffic port={} admin port={} ws port={} a2a={}",
- trafficBoundPort, adminBoundPort, wsBoundPort, a2aGateway != null);
+ // #5411: optional legacy SDK gRPC bridge, started after the core servers so the ingress
+ // it bridges onto is fully up. Pushes share the v2 delivery engine (WAL + retry/DLQ).
+ if (grpcPort >= 0) {
+ grpcBridge = new org.apache.eventmesh.runtime.grpc.EventMeshGrpcServer(
+ runtime.ingress(), grpcPort,
+ (url, body, headers) -> {
+ try {
+ java.net.HttpURLConnection conn = (java.net.HttpURLConnection) new java.net.URL(url).openConnection();
+ conn.setRequestMethod("POST");
+ conn.setDoOutput(true);
+ conn.setConnectTimeout(3_000);
+ conn.setReadTimeout(10_000);
+ for (java.util.Map.Entry<String, String> h : headers.entrySet()) {
+ conn.setRequestProperty(h.getKey(), h.getValue());
+ }
+ try (java.io.OutputStream os = conn.getOutputStream()) {
+ os.write(body);
+ }
+ return conn.getResponseCode();
+ } catch (java.io.IOException e) {
+ throw new IllegalStateException("webhook POST failed: " + e.getMessage(), e);
+ }
+ },
+ event -> event.getData() == null ? new byte[0] : event.getData().toBytes());
+ grpcBridge.start();
+ log.info("legacy gRPC bridge started: port={}", grpcBridge.port());
+ }
+ log.info("EventMeshApplication started: traffic port={} admin port={} ws port={} a2a={} grpc={}",
+ trafficBoundPort, adminBoundPort, wsBoundPort, a2aGateway != null, grpcBridge != null);
}
/** Graceful shutdown: admin → traffic → runtime (flush offsets, release storage). */
@@ -380,6 +425,14 @@
if (wsServer != null) {
wsServer.stop();
}
+ // #5411: stop the gRPC bridge before the runtime so in-flight stream pushes settle.
+ if (grpcBridge != null) {
+ try {
+ grpcBridge.stop();
+ } catch (RuntimeException e) {
+ log.warn("gRPC bridge shutdown: {}", e.toString());
+ }
+ }
// #5405: gateway first (unsubscribes its transport), then its stores.
if (a2aGateway != null) {
try {
@@ -572,6 +625,14 @@
log.info("WebSocket push transport enabled on port {}", wsPort);
}
+ // Legacy SDK gRPC bridge (optional, issue #5411): -Deventmesh.grpc.port=10205 (0 = auto,
+ // omit/negative = disabled - opt-in like the WS port; keeps the 1.x default warm).
+ int grpcPort = Integer.getInteger("eventmesh.grpc.port", -1);
+ if (grpcPort >= 0) {
+ app.withGrpcBridge(grpcPort);
+ log.info("legacy gRPC bridge enabled on port {}", grpcPort);
+ }
+
// A2A gateway (optional, issue #5405): -Deventmesh.a2a.enabled=true, port 10108 by
// default, token via -Deventmesh.a2a.token (open gateway without it — dev mode).
// Task store: local RocksDB under the same data dir by default; the Meta-backed store
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/EventMeshGrpcServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/EventMeshGrpcServer.java
new file mode 100644
index 0000000..93561be
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/EventMeshGrpcServer.java
@@ -0,0 +1,553 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.grpc;
+
+import org.apache.eventmesh.common.protocol.SubscriptionItem;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEventBatch;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.ConsumerServiceGrpc;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.HeartbeatServiceGrpc;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.PublisherServiceGrpc;
+import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
+import org.apache.eventmesh.common.util.JsonUtils;
+import org.apache.eventmesh.runtime.delivery.AckCallback;
+import org.apache.eventmesh.runtime.delivery.CloudEventSerializer;
+import org.apache.eventmesh.runtime.delivery.HttpCaller;
+import org.apache.eventmesh.runtime.delivery.WebHookChannel;
+import org.apache.eventmesh.runtime.ingress.UniIngressService;
+import org.apache.eventmesh.runtime.push.BufferedEvent;
+import org.apache.eventmesh.runtime.push.PushService;
+import org.apache.eventmesh.runtime.subscription.DistributionMode;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
+import io.grpc.stub.StreamObserver;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.google.protobuf.Empty;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * The legacy SDK gRPC protocol compatibility bridge (issue #5411): serves
+ * {@code org.apache.eventmesh.cloudevents.v1.PublisherService/ConsumerService/HeartbeatService} on
+ * the reserved port {@code eventmesh.grpc.port} (1.x default 10205), mapping everything onto the
+ * v2 ingress path - no second messaging engine, the same {@link UniIngressService} the HTTP plane
+ * uses (WAL durability, at-least-once delivery, shared retry/DLQ).
+ *
+ * <p>Opt-in like the WS port: {@code eventmesh.grpc.port} unset / 0 = disabled. Graceful shutdown
+ * with the application lifecycle.</p>
+ *
+ * <p>Mapping summary (compat table in docs/feature/protocols.md):</p>
+ * <ul>
+ * <li>publish / batchPublish / publishOneWay / batchPublishOneWay -> proto CloudEvent mapped to
+ * v2 and persisted via {@code UniIngressService.publish} (topic = proto {@code subject}).</li>
+ * <li>requestReply -> the v2 request/reply correlation ({@code UniIngressService.request});
+ * the reply maps back to a proto CloudEvent. Legacy {@code Response} codes on errors.</li>
+ * <li>subscribe (webhook url) -> a {@link WebHookChannel} push target + v2 subscriptions
+ * (LOAD_BALANCE for the SDK's default CLUSTERING, BROADCAST for BROADCASTING).</li>
+ * <li>subscribeStream (bidi) -> a {@link GrpcStreamChannel} push target pumping the v2 push
+ * pipeline into the stream; client ACKs ride back as stream replies.</li>
+ * <li>heartbeat -> refreshes the {@link GrpcClientRegistry} TTL; the reaper unsubscribes
+ * stale clients (same model as the agent plane).</li>
+ * </ul>
+ */
+@Slf4j
+public class EventMeshGrpcServer {
+
+ /** Attribute key carrying the delivery id on stream pushes (client echoes it to ACK). */
+ public static final String DELIVERY_ID_ATTR = "emdeliveryid";
+
+ private final UniIngressService ingress;
+ private final int port;
+ private final HttpCaller httpCaller;
+ private final CloudEventSerializer serializer;
+
+ private Server server;
+ private GrpcClientRegistry clientRegistry = new GrpcClientRegistry();
+ private ScheduledExecutorService reaper;
+ /** Pending stream deliveries: deliveryId -> ACK callback (fired on client stream ACK). */
+ private final Map<String, AckCallback> pendingStreamAcks = new ConcurrentHashMap<>();
+ /** Per-client stream channel (latest stream wins), for push targeting. */
+ private final Map<String, GrpcStreamChannel> streamChannels = new ConcurrentHashMap<>();
+
+ public EventMeshGrpcServer(UniIngressService ingress, int port,
+ HttpCaller httpCaller, CloudEventSerializer serializer) {
+ this.ingress = ingress;
+ this.port = port;
+ this.httpCaller = httpCaller;
+ this.serializer = serializer;
+ }
+
+ /** Bind and start serving. Returns the actual bound port. */
+ public int start() throws IOException {
+ server = ServerBuilder.forPort(port)
+ .addService(new PublisherImpl())
+ .addService(new ConsumerImpl())
+ .addService(new HeartbeatImpl())
+ .build()
+ .start();
+ // Eviction reaper: same TTL model as the agent plane; unsubscribe dead gRPC clients.
+ clientRegistry.addEvictionListener(clientId -> {
+ ingress.unsubscribeByClient(clientId);
+ GrpcStreamChannel channel = streamChannels.remove(clientId);
+ if (channel != null) {
+ channel.close();
+ }
+ });
+ reaper = Executors.newSingleThreadScheduledExecutor(r -> {
+ Thread t = new Thread(r, "em-grpc-reaper");
+ t.setDaemon(true);
+ return t;
+ });
+ reaper.scheduleAtFixedRate(clientRegistry::reapStale, 30, 30, TimeUnit.SECONDS);
+ int bound = server.getPort();
+ log.info("legacy gRPC bridge started on port {} ({} clients capacity)", bound,
+ InetSocketAddress.createUnresolved("localhost", bound).getPort());
+ return bound;
+ }
+
+ /** Graceful shutdown with the application lifecycle. */
+ public void stop() {
+ if (reaper != null) {
+ reaper.shutdownNow();
+ }
+ if (server != null) {
+ server.shutdown();
+ try {
+ if (!server.awaitTermination(5, TimeUnit.SECONDS)) {
+ server.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ server.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ }
+ log.info("legacy gRPC bridge stopped");
+ }
+
+ public int port() {
+ return server == null ? -1 : server.getPort();
+ }
+
+ // ------------------------------------------------------------------
+ // PublisherService
+ // ------------------------------------------------------------------
+
+ private final class PublisherImpl extends PublisherServiceGrpc.PublisherServiceImplBase {
+
+ @Override
+ public void publish(CloudEvent request, StreamObserver<CloudEvent> responseObserver) {
+ try {
+ String topic = GrpcCloudEventMapper.topicOf(request);
+ if (topic.isEmpty()) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_PROTOCOL_BODY_ERR, "missing subject(topic)"));
+ responseObserver.onCompleted();
+ return;
+ }
+ ingress.publish(topic, GrpcCloudEventMapper.toV2(request))
+ .whenComplete((v, err) -> {
+ if (err != null) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR, err.toString()));
+ } else {
+ responseObserver.onNext(GrpcCloudEventMapper.okResponse());
+ }
+ responseObserver.onCompleted();
+ });
+ } catch (RuntimeException e) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_RUNTIME_ERR, e.toString()));
+ responseObserver.onCompleted();
+ }
+ }
+
+ @Override
+ public void batchPublish(CloudEventBatch request, StreamObserver<CloudEvent> responseObserver) {
+ publishBatchInternal(request, responseObserver, false);
+ }
+
+ @Override
+ public void publishOneWay(CloudEvent request, StreamObserver<Empty> responseObserver) {
+ String topic = GrpcCloudEventMapper.topicOf(request);
+ if (topic.isEmpty()) {
+ responseObserver.onError(new IllegalArgumentException("missing subject(topic)"));
+ return;
+ }
+ ingress.publish(topic, GrpcCloudEventMapper.toV2(request));
+ responseObserver.onNext(Empty.getDefaultInstance());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void batchPublishOneWay(CloudEventBatch request, StreamObserver<Empty> responseObserver) {
+ publishBatchInternal(request, null, true);
+ responseObserver.onNext(Empty.getDefaultInstance());
+ responseObserver.onCompleted();
+ }
+
+ @SuppressWarnings("rawtypes")
+ private void publishBatchInternal(CloudEventBatch request,
+ StreamObserver<CloudEvent> responseObserver, boolean oneWay) {
+ try {
+ if (request.getEventsCount() == 0) {
+ if (!oneWay) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_PROTOCOL_BODY_ERR, "empty batch"));
+ responseObserver.onCompleted();
+ }
+ return;
+ }
+ String topic = GrpcCloudEventMapper.topicOf(request.getEvents(0));
+ if (topic.isEmpty()) {
+ if (!oneWay) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_PROTOCOL_BODY_ERR, "missing subject(topic)"));
+ responseObserver.onCompleted();
+ }
+ return;
+ }
+ List<io.cloudevents.CloudEvent> events = request.getEventsList().stream()
+ .map(GrpcCloudEventMapper::toV2)
+ .collect(java.util.stream.Collectors.toList());
+ CompletableFuture<Void> done = ingress.publishBatch(topic, events);
+ if (oneWay) {
+ return;
+ }
+ done.whenComplete((v, err) -> {
+ if (err != null) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_BATCH_PUBLISH_ERR, err.toString()));
+ } else {
+ responseObserver.onNext(GrpcCloudEventMapper.okResponse());
+ }
+ responseObserver.onCompleted();
+ });
+ } catch (RuntimeException e) {
+ if (!oneWay) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_RUNTIME_ERR, e.toString()));
+ responseObserver.onCompleted();
+ }
+ }
+ }
+
+ @Override
+ public void requestReply(CloudEvent request, StreamObserver<CloudEvent> responseObserver) {
+ try {
+ String topic = GrpcCloudEventMapper.topicOf(request);
+ if (topic.isEmpty()) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_PROTOCOL_BODY_ERR, "missing subject(topic)"));
+ responseObserver.onCompleted();
+ return;
+ }
+ io.cloudevents.CloudEvent reply = ingress.request(
+ topic, GrpcCloudEventMapper.toV2(request), ttlOf(request));
+ responseObserver.onNext(GrpcCloudEventMapper.toProto(reply));
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_REQUEST_REPLY_MSG_ERR, e.toString()));
+ responseObserver.onCompleted();
+ }
+ }
+
+ /** The legacy TTL attribute (ms) drives the request/reply timeout; default 10s like 1.x. */
+ private long ttlOf(CloudEvent request) {
+ String ttl = org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
+ .getTtl(request, "");
+ try {
+ return ttl.isEmpty() ? 10_000L : Long.parseLong(ttl);
+ } catch (NumberFormatException e) {
+ return 10_000L;
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------
+ // ConsumerService
+ // ------------------------------------------------------------------
+
+ private final class ConsumerImpl extends ConsumerServiceGrpc.ConsumerServiceImplBase {
+
+ @Override
+ public void subscribe(CloudEvent request, StreamObserver<CloudEvent> responseObserver) {
+ try {
+ String url = GrpcCloudEventMapper.urlOf(request);
+ if (url.isEmpty() || "grpc_stream".equals(url)) {
+ // A unary subscribe without a webhook URL is invalid - stream subs go through
+ // subscribeStream.
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_SUBSCRIBE_ERR, "missing url (webhook)"));
+ responseObserver.onCompleted();
+ return;
+ }
+ String clientId = subscribeAll(request, new WebHookChannel(
+ url, webhookSecret(), httpCaller, serializer));
+ responseObserver.onNext(GrpcCloudEventMapper.okResponse());
+ responseObserver.onCompleted();
+ log.info("grpc webhook subscriber: clientId={} url={}", clientId, url);
+ } catch (RuntimeException e) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_SUBSCRIBE_ERR, e.toString()));
+ responseObserver.onCompleted();
+ }
+ }
+
+ @Override
+ public void unsubscribe(CloudEvent request, StreamObserver<CloudEvent> responseObserver) {
+ try {
+ String group = GrpcCloudEventMapper.consumerGroupOf(request);
+ String env = org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
+ .getEnv(request, "");
+ String idc = org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
+ .getIdc(request, "");
+ String clientId = clientId(group, env, idc);
+ Set<String> topics = topicsOf(request);
+ int removed = 0;
+ for (String topic : topics) {
+ if (ingress.unsubscribe(topic, clientId)) {
+ removed++;
+ }
+ }
+ clientRegistry.deregister(clientId);
+ GrpcStreamChannel channel = streamChannels.remove(clientId);
+ if (channel != null) {
+ channel.close();
+ }
+ responseObserver.onNext(GrpcCloudEventMapper.okResponse());
+ responseObserver.onCompleted();
+ log.info("grpc unsubscribe: clientId={} topics={} removed={}", clientId, topics, removed);
+ } catch (RuntimeException e) {
+ responseObserver.onNext(GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_UNSUBSCRIBE_ERR, e.toString()));
+ responseObserver.onCompleted();
+ }
+ }
+
+ /**
+ * subscribeStream (bidi): the first inbound message carries the subscription (same
+ * envelope as unary subscribe); every later message is either a reply (SUB_REPLY_MESSAGE)
+ * or an ACK ({@link #DELIVERY_ID_ATTR} echoed). Push goes through the v2 dispatcher onto
+ * the {@link GrpcStreamChannel}.
+ */
+ @Override
+ public StreamObserver<CloudEvent> subscribeStream(StreamObserver<CloudEvent> responseObserver) {
+ return new StreamObserver<CloudEvent>() {
+
+ private String clientId;
+
+ @Override
+ public void onNext(CloudEvent message) {
+ boolean isAck = message.getAttributesMap().containsKey(DELIVERY_ID_ATTR)
+ && !message.getAttributesMap().containsKey(ProtocolKey.SUB_MESSAGE_TYPE);
+ if (isAck) {
+ ackFromClient(message);
+ return;
+ }
+ if (clientId == null) {
+ // first message = the subscription envelope
+ GrpcStreamChannel channel = new GrpcStreamChannel(responseObserver);
+ clientId = subscribeAll(message, channel);
+ streamChannels.put(clientId, channel);
+ // acknowledge the subscription itself (1.x sent an ack envelope)
+ synchronized (responseObserver) {
+ responseObserver.onNext(GrpcCloudEventMapper.okResponse());
+ }
+ } else {
+ // later non-ack messages: a reply to a request/reply event routed over
+ // the stream - correlate via emcorrelationid if present.
+ String correlationId = attr(message, UniIngressService.EXT_CORRELATION_ID);
+ if (correlationId != null && !correlationId.isEmpty()) {
+ ingress.reply(correlationId, GrpcCloudEventMapper.toV2(message));
+ }
+ }
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ closeStream();
+ log.info("grpc subscribeStream error: {}", t.toString());
+ }
+
+ @Override
+ public void onCompleted() {
+ closeStream();
+ }
+
+ private void closeStream() {
+ if (clientId != null) {
+ GrpcStreamChannel channel = streamChannels.remove(clientId);
+ if (channel != null) {
+ channel.close();
+ }
+ }
+ }
+ };
+ }
+
+ private String attr(CloudEvent message, String key) {
+ CloudEvent.CloudEventAttributeValue v = message.getAttributesMap().get(key);
+ return v == null ? null : v.getCeString();
+ }
+
+ /** Fire the pending stream ACK when the client echoes the delivery id back. */
+ private void ackFromClient(CloudEvent message) {
+ String deliveryId = attr(message, DELIVERY_ID_ATTR);
+ AckCallback callback = deliveryId == null ? null : pendingStreamAcks.remove(deliveryId);
+ if (callback != null) {
+ callback.ack();
+ } else {
+ // Fallback: long-polling style ACK via the ingress (delivery ids are shared).
+ ingress.ack(deliveryId);
+ }
+ }
+
+ /**
+ * Shared subscribe path: parse the subscription items, register the client + push channel,
+ * map each topic to a v2 subscription. Returns the derived clientId.
+ */
+ private String subscribeAll(CloudEvent request,
+ org.apache.eventmesh.runtime.delivery.PushChannel channel) {
+ String group = GrpcCloudEventMapper.consumerGroupOf(request);
+ String env = org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
+ .getEnv(request, "");
+ String idc = org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
+ .getIdc(request, "");
+ String clientId = clientId(group, env, idc);
+ clientRegistry.register(clientId, group);
+ ingress.registerChannel(clientId, channel);
+ for (SubscriptionItem item : itemsOf(request)) {
+ DistributionMode mode = item != null && item.getMode() != null
+ && "BROADCASTING".equals(item.getMode().getMode())
+ ? DistributionMode.BROADCAST : DistributionMode.LOAD_BALANCE;
+ ingress.subscribe(item.getTopic(), clientId, mode, null);
+ }
+ return clientId;
+ }
+
+ /** The 1.x clientId derivation: consumerGroup + env + idc (issue #5411 item 3). */
+ private String clientId(String group, String env, String idc) {
+ return group + "-" + env + "-" + idc;
+ }
+
+ private Set<String> topicsOf(CloudEvent request) {
+ Set<String> topics = new HashSet<>();
+ for (SubscriptionItem item : itemsOf(request)) {
+ topics.add(item.getTopic());
+ }
+ return topics;
+ }
+
+ private List<SubscriptionItem> itemsOf(CloudEvent request) {
+ String json = request.getTextData();
+ if (json == null || json.isEmpty()) {
+ return java.util.Collections.emptyList();
+ }
+ return JsonUtils.parseTypeReferenceObject(json,
+ new TypeReference<Set<SubscriptionItem>>() {
+ }) == null ? java.util.Collections.emptyList()
+ : new java.util.ArrayList<>(JsonUtils.parseTypeReferenceObject(json,
+ new TypeReference<Set<SubscriptionItem>>() {
+ }));
+ }
+ }
+
+ // ------------------------------------------------------------------
+ // HeartbeatService
+ // ------------------------------------------------------------------
+
+ private final class HeartbeatImpl extends HeartbeatServiceGrpc.HeartbeatServiceImplBase {
+
+ @Override
+ public void heartbeat(CloudEvent request, StreamObserver<CloudEvent> responseObserver) {
+ String group = GrpcCloudEventMapper.consumerGroupOf(request);
+ String env = org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
+ .getEnv(request, "");
+ String idc = org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils
+ .getIdc(request, "");
+ String clientId = new ConsumerImpl().clientId(group, env, idc);
+ boolean known = clientRegistry.heartbeat(clientId);
+ if (!known) {
+ // 1.x answered CLIENT_RESUBSCRIBE for unknown clients - the SDK resubscribes.
+ responseObserver.onNext(GrpcCloudEventMapper.response(
+ StatusCode.CLIENT_RESUBSCRIBE.getRetCode(),
+ StatusCode.CLIENT_RESUBSCRIBE.getErrMsg()));
+ responseObserver.onCompleted();
+ return;
+ }
+ responseObserver.onNext(GrpcCloudEventMapper.okResponse());
+ responseObserver.onCompleted();
+ }
+ }
+
+ /** Webhook signing secret (fixed default; per-URL secrets are a follow-up like 1.x). */
+ private String webhookSecret() {
+ return "eventmesh-grpc-webhook";
+ }
+
+ // ---- test accessors ----
+
+ GrpcClientRegistry registryForTest() {
+ return clientRegistry;
+ }
+
+ Map<String, AckCallback> pendingStreamAcksForTest() {
+ return pendingStreamAcks;
+ }
+
+ PushService pushServiceForTest() {
+ return ingress.getPushService();
+ }
+
+ List<BufferedEvent> pollForTest(String clientId, int max, long timeoutMs) {
+ return ingress.poll(clientId, max, timeoutMs);
+ }
+
+ Map<String, GrpcStreamChannel> streamChannelsForTest() {
+ return streamChannels;
+ }
+
+ Map<String, AckCallback> streamAckRegistry() {
+ return pendingStreamAcks;
+ }
+
+ @SuppressWarnings("unused")
+ private Map<String, String> diagnosticSnapshot() {
+ Map<String, String> snap = new HashMap<>();
+ snap.put("clients", Integer.toString(clientRegistry.size()));
+ snap.put("streamChannels", Integer.toString(streamChannels.size()));
+ return snap;
+ }
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcClientRegistry.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcClientRegistry.java
new file mode 100644
index 0000000..bdc58e4
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcClientRegistry.java
@@ -0,0 +1,147 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.grpc;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.function.LongSupplier;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * In-memory registration + TTL table for gRPC clients (issue #5411 item 4/5): the 1.x heartbeat
+ * model - a client registers on subscribe, refreshes on every
+ * {@code HeartbeatService.heartbeat} call, and a reaper evicts clients whose last heartbeat is
+ * older than the TTL (default 3x the SDK's 30s heartbeat interval, matching the 1.x slack).
+ *
+ * <p>Eviction drops the client's v2 subscriptions so a dead gRPC consumer does not pin delivery
+ * targets forever (the dispatcher would keep buffering into its PushService slot).</p>
+ */
+@Slf4j
+public class GrpcClientRegistry {
+
+ /** Default TTL: 3x the SDK heartbeat interval (30s) - 1.x used the same slack. */
+ public static final long DEFAULT_TTL_MS = 90_000L;
+
+ /** Called with the clientId of every evicted client (unsubscribe hook, may be null in tests). */
+ public interface EvictionListener {
+
+ void onEvicted(String clientId);
+ }
+
+ private static final class Entry {
+
+ final String clientId;
+ final String consumerGroup;
+ volatile long lastHeartbeatMs;
+
+ Entry(String clientId, String consumerGroup, long now) {
+ this.clientId = clientId;
+ this.consumerGroup = consumerGroup;
+ this.lastHeartbeatMs = now;
+ }
+ }
+
+ private final long ttlMs;
+ private final LongSupplier clock;
+ private final Map<String, Entry> clients = new ConcurrentHashMap<>();
+ private final List<EvictionListener> listeners = new CopyOnWriteArrayList<>();
+
+ public GrpcClientRegistry() {
+ this(DEFAULT_TTL_MS, System::currentTimeMillis);
+ }
+
+ public GrpcClientRegistry(long ttlMs, LongSupplier clock) {
+ this.ttlMs = ttlMs;
+ this.clock = clock;
+ }
+
+ public void addEvictionListener(EvictionListener listener) {
+ listeners.add(listener);
+ }
+
+ /** Register (or refresh) a client. Idempotent. */
+ public void register(String clientId, String consumerGroup) {
+ long now = clock.getAsLong();
+ clients.compute(clientId, (id, existing) -> {
+ if (existing == null) {
+ return new Entry(id, consumerGroup, now);
+ }
+ existing.lastHeartbeatMs = now;
+ return existing;
+ });
+ }
+
+ /** Refresh the TTL of a registered client. Returns false for unknown clients. */
+ public boolean heartbeat(String clientId) {
+ Entry entry = clients.get(clientId);
+ if (entry == null) {
+ return false;
+ }
+ entry.lastHeartbeatMs = clock.getAsLong();
+ return true;
+ }
+
+ /** Deregister a client (unsubscribe / stream close). */
+ public void deregister(String clientId) {
+ clients.remove(clientId);
+ }
+
+ public boolean isRegistered(String clientId) {
+ return clients.containsKey(clientId);
+ }
+
+ public int size() {
+ return clients.size();
+ }
+
+ /**
+ * Evict clients whose last heartbeat is older than the TTL.
+ *
+ * @return the number of clients evicted (each fires the eviction listeners)
+ */
+ public int reapStale() {
+ long now = clock.getAsLong();
+ int evicted = 0;
+ for (Map.Entry<String, Entry> e : clients.entrySet()) {
+ Entry entry = e.getValue();
+ if (now - entry.lastHeartbeatMs > ttlMs) {
+ if (clients.remove(e.getKey(), entry)) {
+ evicted++;
+ log.info("grpc client evicted (heartbeat TTL): clientId={} group={}",
+ entry.clientId, entry.consumerGroup);
+ for (EvictionListener listener : listeners) {
+ try {
+ listener.onEvicted(entry.clientId);
+ } catch (RuntimeException ex) {
+ log.warn("eviction listener failed for {}: {}", entry.clientId, ex.toString());
+ }
+ }
+ }
+ }
+ }
+ return evicted;
+ }
+
+ long lastHeartbeatMs(String clientId) {
+ Entry entry = clients.get(clientId);
+ return entry == null ? -1L : entry.lastHeartbeatMs;
+ }
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcCloudEventMapper.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcCloudEventMapper.java
new file mode 100644
index 0000000..b7ce4f5
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcCloudEventMapper.java
@@ -0,0 +1,212 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.grpc;
+
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent.CloudEventAttributeValue;
+import org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
+import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.time.OffsetDateTime;
+import java.util.Map;
+
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Maps the legacy SDK gRPC proto {@link CloudEvent} (org.apache.eventmesh.cloudevents.v1) to the v2
+ * runtime's {@link io.cloudevents.CloudEvent} and back (issue #5411).
+ *
+ * <p>Inbound (proto -> v2): the proto's well-known attribute keys (subject carries the topic; the
+ * {@code ProtocolKey.*} entries become CloudEvents extensions) and the oneof data field
+ * (textData / binaryData / protoData) become the event data bytes. Outbound (v2 -> proto) is the
+ * inverse, used by the subscribeStream push path so the legacy
+ * {@code EventMeshGrpcConsumer} can deserialize what it receives
+ * ({@code EventMeshCloudEventBuilder#buildMessageFromEventMeshCloudEvent} with the
+ * EVENT_MESH_MESSAGE protocol type reads textData + subject/seqnum/uniqueid attributes).</p>
+ *
+ * <p>Pure functions, no runtime state - hermetically unit-testable.</p>
+ */
+public final class GrpcCloudEventMapper {
+
+ private GrpcCloudEventMapper() {
+ }
+
+ /**
+ * proto {@link CloudEvent} -> v2 {@link io.cloudevents.CloudEvent}.
+ *
+ * <p>The proto's {@code subject} attribute carries the destination topic (see
+ * {@code EventMeshCloudEventBuilder#switchEventMeshMessage2EventMeshCloudEvent}); it is mapped
+ * onto the CloudEvents {@code subject} attribute so the v2 pipeline keeps topic routing intact.
+ * All other attributes become CloudEvents extensions (string-typed; numeric/boolean proto
+ * attribute values are flattened to their string form).</p>
+ */
+ public static io.cloudevents.CloudEvent toV2(CloudEvent proto) {
+ CloudEventBuilder builder = CloudEventBuilder.v1()
+ .withId(proto.getId())
+ .withSource(URI.create(proto.getSource().isEmpty() ? "/" : proto.getSource()))
+ .withType(proto.getType().isEmpty() ? "org.apache.eventmesh" : proto.getType());
+ if (proto.hasTextData()) {
+ builder.withData(proto.getTextData().getBytes(StandardCharsets.UTF_8));
+ } else if (proto.hasBinaryData()) {
+ builder.withData(proto.getBinaryData().toByteArray());
+ } else if (proto.hasProtoData()) {
+ builder.withData(proto.getProtoData().toByteArray());
+ }
+ for (Map.Entry<String, CloudEventAttributeValue> attr : proto.getAttributesMap().entrySet()) {
+ String key = attr.getKey();
+ String value = stringify(attr.getValue());
+ if (value == null) {
+ continue;
+ }
+ if (ProtocolKey.SUBJECT.equals(key)) {
+ builder.withSubject(value);
+ } else {
+ builder.withExtension(key, value);
+ }
+ }
+ return builder.build();
+ }
+
+ /**
+ * v2 {@link io.cloudevents.CloudEvent} -> proto {@link CloudEvent}, for the subscribeStream
+ * push path. Mirrors what the legacy 1.x runtime sent: id/source/specversion/type top-level,
+ * every attribute + extension in the attributes map, and the event data as
+ * {@code textData} (the legacy SDK's EventMeshMessage deserializer reads textData; binary
+ * payloads are surfaced through textData as UTF-8 like 1.x did for String contents).
+ */
+ public static CloudEvent toProto(io.cloudevents.CloudEvent event) {
+ CloudEvent.Builder builder = CloudEvent.newBuilder()
+ .setId(event.getId() == null ? "" : event.getId())
+ .setSource(event.getSource() == null ? "/" : event.getSource().toString())
+ .setSpecVersion(event.getSpecVersion() == null ? "1.0" : event.getSpecVersion().toString())
+ .setType(event.getType() == null ? "org.apache.eventmesh" : event.getType());
+ if (event.getSubject() != null) {
+ builder.putAttributes(ProtocolKey.SUBJECT, CloudEventAttributeValue.newBuilder()
+ .setCeString(event.getSubject()).build());
+ }
+ for (String name : event.getAttributeNames()) {
+ Object v = event.getAttribute(name);
+ if (v != null) {
+ putAttr(builder, name, v.toString());
+ }
+ }
+ for (String name : event.getExtensionNames()) {
+ Object v = event.getExtension(name);
+ if (v != null) {
+ putAttr(builder, name, v.toString());
+ }
+ }
+ if (event.getData() != null) {
+ byte[] data = event.getData().toBytes();
+ if (data.length > 0) {
+ builder.setTextData(new String(data, StandardCharsets.UTF_8));
+ }
+ }
+ return builder.build();
+ }
+
+ /**
+ * Build a legacy {@link CloudEvent} response envelope: the 1.x contract answers publish /
+ * subscribe / unsubscribe / heartbeat calls with a CloudEvent whose attributes carry
+ * {@code statuscode} / {@code responsemessage} / {@code time} (see {@code StatusCode} +
+ * {@code EventMeshCloudEventUtils#getResponseCode}). The SDK parses exactly these three keys.
+ */
+ public static CloudEvent response(String retCode, String message) {
+ return CloudEvent.newBuilder()
+ .setId("resp-" + java.util.UUID.randomUUID())
+ .setSource("/")
+ .setSpecVersion("1.0")
+ .setType("org.apache.eventmesh")
+ .putAttributes(ProtocolKey.GRPC_RESPONSE_CODE,
+ CloudEventAttributeValue.newBuilder().setCeString(retCode).build())
+ .putAttributes(ProtocolKey.GRPC_RESPONSE_MESSAGE,
+ CloudEventAttributeValue.newBuilder().setCeString(message).build())
+ .putAttributes(ProtocolKey.GRPC_RESPONSE_TIME,
+ CloudEventAttributeValue.newBuilder()
+ .setCeTimestamp(com.google.protobuf.Timestamp.newBuilder()
+ .setSeconds(System.currentTimeMillis() / 1000)
+ .build())
+ .build())
+ .build();
+ }
+
+ /** Legacy success envelope (StatusCode SUCCESS = "0"). */
+ public static CloudEvent okResponse() {
+ return response(org.apache.eventmesh.common.protocol.grpc.common.StatusCode.SUCCESS.getRetCode(),
+ org.apache.eventmesh.common.protocol.grpc.common.StatusCode.SUCCESS.getErrMsg());
+ }
+
+ /** Legacy error envelope for the given status code (+ detail appended to the stock message). */
+ public static CloudEvent errorResponse(
+ org.apache.eventmesh.common.protocol.grpc.common.StatusCode code, String detail) {
+ return response(code.getRetCode(), code.getErrMsg() + (detail == null ? "" : detail));
+ }
+
+ private static void putAttr(CloudEvent.Builder builder, String key, String value) {
+ if (key == null || value == null) {
+ return;
+ }
+ builder.putAttributes(key, CloudEventAttributeValue.newBuilder().setCeString(value).build());
+ }
+
+ private static String stringify(CloudEventAttributeValue value) {
+ if (value == null) {
+ return null;
+ }
+ if (value.hasCeString()) {
+ return value.getCeString();
+ }
+ if (value.hasCeInteger()) {
+ return Integer.toString(value.getCeInteger());
+ }
+ if (value.hasCeBoolean()) {
+ return Boolean.toString(value.getCeBoolean());
+ }
+ if (value.hasCeUri() || value.hasCeUriRef()) {
+ return value.hasCeUri() ? value.getCeUri() : value.getCeUriRef();
+ }
+ if (value.hasCeBytes()) {
+ return value.getCeBytes().toString(StandardCharsets.UTF_8);
+ }
+ if (value.hasCeTimestamp()) {
+ OffsetDateTime t = OffsetDateTime.ofInstant(
+ java.time.Instant.ofEpochSecond(value.getCeTimestamp().getSeconds(),
+ value.getCeTimestamp().getNanos()),
+ java.time.ZoneOffset.UTC);
+ return t.toString();
+ }
+ return null;
+ }
+
+ /** The destination topic of a publish/subscribe proto event ({@code subject} attribute). */
+ public static String topicOf(CloudEvent proto) {
+ return EventMeshCloudEventUtils.getSubject(proto, "");
+ }
+
+ /** The consumer group of a subscribe/heartbeat proto event. */
+ public static String consumerGroupOf(CloudEvent proto) {
+ return EventMeshCloudEventUtils.getConsumerGroup(proto, "");
+ }
+
+ /** The webhook URL of a subscribe proto event (empty for stream subscriptions). */
+ public static String urlOf(CloudEvent proto) {
+ return EventMeshCloudEventUtils.getURL(proto, "");
+ }
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcStreamChannel.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcStreamChannel.java
new file mode 100644
index 0000000..734c314
--- /dev/null
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/grpc/GrpcStreamChannel.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.grpc;
+
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+import org.apache.eventmesh.runtime.delivery.AckCallback;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import io.grpc.stub.StreamObserver;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * The subscribeStream bridge (issue #5411 item 3): adapts one client-held bidi gRPC stream into a
+ * {@link org.apache.eventmesh.runtime.delivery.PushChannel}, so the v2 delivery pipeline
+ * (ReliableDispatcher -> channel.deliver) pushes straight into the stream, and the SDK's
+ * client-side ACK (a reply message sent back on the same stream) completes the delivery via
+ * {@link #ack(String)}.
+ *
+ * <p>This mirrors the WebSocket/SSE transport contract (buffered + ACK-tracked at-least-once,
+ * shared retry/DLQ) - it is the "same dispatcher the SSE path uses", just with a gRPC
+ * {@link StreamObserver} underneath.</p>
+ */
+@Slf4j
+public class GrpcStreamChannel implements org.apache.eventmesh.runtime.delivery.PushChannel {
+
+ private final StreamObserver<CloudEvent> sink;
+ private final AtomicBoolean open = new AtomicBoolean(true);
+
+ public GrpcStreamChannel(StreamObserver<CloudEvent> sink) {
+ this.sink = sink;
+ }
+
+ @Override
+ public void deliver(String deliveryId, EventMeshFrame event, AckCallback callback) {
+ if (!open.get()) {
+ callback.nack(new IllegalStateException("grpc stream closed"));
+ return;
+ }
+ try {
+ // Egress: Frame -> v2 CloudEvent -> legacy proto CloudEvent. The delivery id rides as
+ // the seqnum attribute: the legacy SDK replies with the SAME attributes (its reply
+ // builder copies the request's attribute map), so the ack path can correlate back.
+ io.cloudevents.CloudEvent v2 = event.toCloudEvent();
+ CloudEvent proto = GrpcCloudEventMapper.toProto(v2);
+ proto = CloudEvent.newBuilder(proto)
+ .putAttributes("emdeliveryid",
+ CloudEvent.CloudEventAttributeValue.newBuilder().setCeString(deliveryId).build())
+ .build();
+ synchronized (sink) {
+ sink.onNext(proto);
+ }
+ // NOTE: the ACK fires from the stream's inbound side (GrpcSubscriber reply or the
+ // dedicated ack attribute) - see EventMeshGrpcServer.ackFromClient. Until then the
+ // callback stays pending in the dispatcher's tracker (at-least-once: unACKed
+ // deliveries redeliver after ackTimeoutMs).
+ } catch (RuntimeException e) {
+ open.set(false);
+ log.warn("grpc stream push failed (delivery={}): {}", deliveryId, e.toString());
+ callback.nack(e);
+ }
+ }
+
+ /** Complete a pending delivery when the client ACKs on the stream. */
+ public boolean ack(String deliveryId) {
+ return true; // correlation handled by the server's AckCallback registry
+ }
+
+ public boolean isOpen() {
+ return open.get();
+ }
+
+ /** Mark the stream dead (client cancelled / errored). Buffered events will redeliver. */
+ public void close() {
+ open.set(false);
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/grpc/GrpcCloudEventMapperTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/grpc/GrpcCloudEventMapperTest.java
new file mode 100644
index 0000000..cb7d3f3
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/grpc/GrpcCloudEventMapperTest.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.grpc;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.eventmesh.common.protocol.grpc.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.protocol.grpc.common.EventMeshCloudEventUtils;
+import org.apache.eventmesh.common.protocol.grpc.common.ProtocolKey;
+import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
+
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Hermetic unit tests for the proto <-> v2 CloudEvent mapping (issue #5411 item 7): every
+ * attribute flavor the legacy SDK sends must round-trip, and the response envelope must carry the
+ * exact three keys {@code EventMeshCloudEventUtils} parses on the client side.
+ */
+class GrpcCloudEventMapperTest {
+
+ @Test
+ void protoToV2MapsTopicSubjectAndData() {
+ CloudEvent proto = CloudEvent.newBuilder()
+ .setId("p-1")
+ .setSource("/")
+ .setSpecVersion("1.0")
+ .setType("org.apache.eventmesh")
+ .setTextData("hello-grpc")
+ .putAttributes(ProtocolKey.SUBJECT,
+ CloudEvent.CloudEventAttributeValue.newBuilder().setCeString("orders").build())
+ .putAttributes(ProtocolKey.SEQ_NUM,
+ CloudEvent.CloudEventAttributeValue.newBuilder().setCeString("seq-1").build())
+ .putAttributes(ProtocolKey.TTL,
+ CloudEvent.CloudEventAttributeValue.newBuilder().setCeString("10000").build())
+ .putAttributes(ProtocolKey.CONSUMERGROUP,
+ CloudEvent.CloudEventAttributeValue.newBuilder().setCeInteger(4).build())
+ .build();
+
+ io.cloudevents.CloudEvent v2 = GrpcCloudEventMapper.toV2(proto);
+ assertEquals("p-1", v2.getId());
+ assertEquals("orders", v2.getSubject(), "subject attribute (the topic) must map to CE subject");
+ assertEquals("hello-grpc", new String(v2.getData().toBytes(), StandardCharsets.UTF_8));
+ assertEquals("seq-1", v2.getExtension("seqnum"));
+ assertEquals("10000", v2.getExtension("ttl"));
+ assertEquals("4", v2.getExtension("consumergroup"), "integer attr flattens to string");
+ }
+
+ @Test
+ void protoToV2BinaryAndProtoData() {
+ CloudEvent binary = CloudEvent.newBuilder().setId("b-1").setSource("/").setType("t")
+ .setBinaryData(com.google.protobuf.ByteString.copyFrom(new byte[] {1, 2, 3})).build();
+ assertEquals(3, GrpcCloudEventMapper.toV2(binary).getData().toBytes().length);
+
+ CloudEvent protoData = CloudEvent.newBuilder().setId("pd-1").setSource("/").setType("t")
+ .setProtoData(com.google.protobuf.Any.getDefaultInstance()).build();
+ assertNotNull(GrpcCloudEventMapper.toV2(protoData).getData());
+ }
+
+ @Test
+ void v2ToProtoRoundTripsSubjectAndTextData() {
+ io.cloudevents.CloudEvent v2 = CloudEventBuilder.v1()
+ .withId("v-1").withSource(URI.create("/")).withType("org.apache.eventmesh")
+ .withSubject("orders")
+ .withData("payload".getBytes(StandardCharsets.UTF_8))
+ .withExtension("seqnum", "s-9")
+ .build();
+ CloudEvent proto = GrpcCloudEventMapper.toProto(v2);
+ assertEquals("v-1", proto.getId());
+ assertEquals("orders", EventMeshCloudEventUtils.getSubject(proto));
+ assertEquals("payload", proto.getTextData());
+ assertEquals("s-9", EventMeshCloudEventUtils.getSeqNum(proto));
+ }
+
+ @Test
+ void responseEnvelopeCarriesTheThreeSdkKeys() {
+ CloudEvent ok = GrpcCloudEventMapper.okResponse();
+ assertEquals(StatusCode.SUCCESS.getRetCode(), EventMeshCloudEventUtils.getResponseCode(ok));
+ assertEquals(StatusCode.SUCCESS.getErrMsg(), EventMeshCloudEventUtils.getResponseMessage(ok));
+ assertNotNull(EventMeshCloudEventUtils.getResponseTime(ok));
+
+ CloudEvent err = GrpcCloudEventMapper.errorResponse(
+ StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR, "boom");
+ assertEquals(StatusCode.EVENTMESH_SEND_ASYNC_MSG_ERR.getRetCode(),
+ EventMeshCloudEventUtils.getResponseCode(err));
+ assertTrue(EventMeshCloudEventUtils.getResponseMessage(err).contains("boom"));
+ }
+
+ @Test
+ void helpersExtractTopicGroupUrl() {
+ CloudEvent sub = CloudEvent.newBuilder().setId("sub-1").setSource("/").setType("t")
+ .putAttributes(ProtocolKey.SUBJECT,
+ CloudEvent.CloudEventAttributeValue.newBuilder().setCeString("t1").build())
+ .putAttributes(ProtocolKey.CONSUMERGROUP,
+ CloudEvent.CloudEventAttributeValue.newBuilder().setCeString("g1").build())
+ .putAttributes(ProtocolKey.URL,
+ CloudEvent.CloudEventAttributeValue.newBuilder().setCeString("http://cb").build())
+ .build();
+ assertEquals("t1", GrpcCloudEventMapper.topicOf(sub));
+ assertEquals("g1", GrpcCloudEventMapper.consumerGroupOf(sub));
+ assertEquals("http://cb", GrpcCloudEventMapper.urlOf(sub));
+
+ CloudEvent bare = CloudEvent.newBuilder().setId("x").setSource("/").setType("t").build();
+ assertEquals("", GrpcCloudEventMapper.topicOf(bare));
+ assertNull(null); // readability anchor
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/grpc/GrpcLegacyBridgeIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/grpc/GrpcLegacyBridgeIntegrationTest.java
new file mode 100644
index 0000000..cab9bea
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/grpc/GrpcLegacyBridgeIntegrationTest.java
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.grpc;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.eventmesh.client.grpc.config.EventMeshGrpcClientConfig;
+import org.apache.eventmesh.client.grpc.consumer.EventMeshGrpcConsumer;
+import org.apache.eventmesh.client.grpc.producer.EventMeshGrpcProducer;
+import org.apache.eventmesh.common.protocol.SubscriptionItem;
+import org.apache.eventmesh.common.protocol.SubscriptionMode;
+import org.apache.eventmesh.common.protocol.SubscriptionType;
+import org.apache.eventmesh.common.protocol.grpc.common.Response;
+import org.apache.eventmesh.common.protocol.grpc.common.StatusCode;
+import org.apache.eventmesh.runtime.boot.UniRuntime;
+import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
+
+import java.net.URI;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+/**
+ * Integration test booting the legacy gRPC bridge against the in-memory storage backend and
+ * driving it with the REAL legacy SDK clients (issue #5411 item 7: "the SDK is its own conformance
+ * suite") - {@link EventMeshGrpcProducer} publish / batch publish, and the
+ * {@link EventMeshGrpcConsumer} stream subscription receiving what was published.
+ *
+ * <p>Always on (no broker needed): the memory storage plugin boots in-process, exactly like
+ * {@code MemoryStorageE2EIntegrationTest}.</p>
+ */
+class GrpcLegacyBridgeIntegrationTest {
+
+ private UniRuntime runtime;
+ private EventMeshGrpcServer grpc;
+ private int port;
+
+ @BeforeEach
+ void boot() throws Exception {
+ runtime = new UniRuntime(new InMemoryStorage(), new InMemoryOffsetStore(), 50L, 200L, 100, 500L);
+ runtime.withStorageConfig(new java.util.Properties());
+ runtime.start();
+
+ grpc = new EventMeshGrpcServer(runtime.ingress(), 0,
+ (url, body, headers) -> 200,
+ event -> event.getData() == null ? new byte[0] : event.getData().toBytes());
+ port = grpc.start();
+ }
+
+ @AfterEach
+ void shutdown() {
+ if (grpc != null) {
+ grpc.stop();
+ }
+ if (runtime != null) {
+ runtime.shutdown();
+ }
+ }
+
+ private EventMeshGrpcClientConfig config() {
+ return EventMeshGrpcClientConfig.builder()
+ .serverAddr("127.0.0.1")
+ .serverPort(port)
+ .producerGroup("TestProducerGroup")
+ .consumerGroup("TestConsumerGroup")
+ .env("it-env")
+ .idc("it-idc")
+ .build();
+ }
+
+ @Test
+ void sdkPublishReceivesSuccessEnvelope() throws Exception {
+ try (EventMeshGrpcProducer producer = new EventMeshGrpcProducer(config())) {
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("it-1").withSource(URI.create("/it")).withType("it.event")
+ .withSubject("grpc-it-topic")
+ .withData("ping".getBytes(java.nio.charset.StandardCharsets.UTF_8))
+ .build();
+ Response resp = producer.publish(event);
+ assertNotNull(resp);
+ assertEquals(StatusCode.SUCCESS.getRetCode(), resp.getRespCode(),
+ "publish should answer the legacy SUCCESS envelope, got: " + resp);
+ }
+ }
+
+ @Test
+ void sdkBatchPublishReceivesSuccessEnvelope() throws Exception {
+ try (EventMeshGrpcProducer producer = new EventMeshGrpcProducer(config())) {
+ CloudEvent e1 = CloudEventBuilder.v1().withId("b-1").withSource(URI.create("/it"))
+ .withType("it.event").withSubject("grpc-it-batch")
+ .withData("1".getBytes(java.nio.charset.StandardCharsets.UTF_8)).build();
+ CloudEvent e2 = CloudEventBuilder.v1().withId("b-2").withSource(URI.create("/it"))
+ .withType("it.event").withSubject("grpc-it-batch")
+ .withData("2".getBytes(java.nio.charset.StandardCharsets.UTF_8)).build();
+ Response resp = producer.publish(List.of(e1, e2));
+ assertNotNull(resp);
+ assertEquals(StatusCode.SUCCESS.getRetCode(), resp.getRespCode(),
+ "batch publish should answer SUCCESS, got: " + resp);
+ }
+ }
+
+ @Test
+ void sdkStreamSubscriptionReceivesPublishedEvent() throws Exception {
+ String topic = "grpc-it-stream";
+ List<Object> received = new CopyOnWriteArrayList<>();
+ CountDownLatch got = new CountDownLatch(1);
+
+ EventMeshGrpcConsumer consumer = new EventMeshGrpcConsumer(config());
+ consumer.init();
+ consumer.registerListener(new org.apache.eventmesh.client.grpc.consumer.ReceiveMsgHook<Object>() {
+
+ @Override
+ public java.util.Optional<Object> handle(Object msg) {
+ received.add(msg);
+ got.countDown();
+ return java.util.Optional.empty();
+ }
+
+ @Override
+ public org.apache.eventmesh.common.enums.EventMeshProtocolType getProtocolType() {
+ return org.apache.eventmesh.common.enums.EventMeshProtocolType.EVENT_MESH_MESSAGE;
+ }
+ });
+ consumer.subscribe(List.of(new SubscriptionItem(topic,
+ SubscriptionMode.CLUSTERING, SubscriptionType.ASYNC)));
+
+ try (EventMeshGrpcProducer producer = new EventMeshGrpcProducer(config())) {
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId("s-1").withSource(URI.create("/it")).withType("it.event")
+ .withSubject(topic)
+ .withData("stream-payload".getBytes(java.nio.charset.StandardCharsets.UTF_8))
+ .build();
+ Response resp = producer.publish(event);
+ assertNotNull(resp);
+ assertEquals(StatusCode.SUCCESS.getRetCode(), resp.getRespCode());
+ }
+
+ assertTrue(got.await(15, TimeUnit.SECONDS),
+ "stream subscriber did not receive the event within 15s");
+ assertEquals(1, received.size());
+ consumer.close();
+ }
+
+ /** Same in-memory storage fixture the other runtime ITs use (queue-per-topic). */
+ private static final class InMemoryStorage implements org.apache.eventmesh.api.storage.MeshStoragePlugin {
+
+ private final java.util.concurrent.ConcurrentHashMap<String,
+ java.util.Queue<io.cloudevents.CloudEvent>> queues = new java.util.concurrent.ConcurrentHashMap<>();
+
+ @Override
+ public void init(java.util.Properties properties) {
+ }
+
+ @Override
+ public void send(String topic, org.apache.eventmesh.common.wire.EventMeshFrame frame,
+ org.apache.eventmesh.api.SendCallback callback) {
+ io.cloudevents.CloudEvent event = frame.toCloudEvent();
+ queues.computeIfAbsent(topic, k -> new java.util.concurrent.ConcurrentLinkedQueue<>()).offer(event);
+ org.apache.eventmesh.api.SendResult r = new org.apache.eventmesh.api.SendResult();
+ r.setMessageId(event.getId());
+ r.setTopic(topic);
+ callback.onSuccess(r);
+ }
+
+ @Override
+ public java.util.List<org.apache.eventmesh.common.wire.EventMeshFrame> poll(
+ String topic, int partition, long startOffset, int maxEvents, long timeoutMs) {
+ java.util.Queue<io.cloudevents.CloudEvent> q = queues.get(topic);
+ if (q == null) {
+ return new java.util.ArrayList<>();
+ }
+ java.util.List<org.apache.eventmesh.common.wire.EventMeshFrame> out = new java.util.ArrayList<>();
+ io.cloudevents.CloudEvent e;
+ while (out.size() < maxEvents && (e = q.poll()) != null) {
+ out.add(org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(e));
+ }
+ return out;
+ }
+
+ @Override
+ public void assignPartitions(String topic, java.util.List<Integer> partitions) {
+ }
+
+ @Override
+ public void commitOffset(String topic, int partition, long offset) {
+ }
+
+ @Override
+ public boolean isStarted() {
+ return true;
+ }
+
+ @Override
+ public boolean isClosed() {
+ return false;
+ }
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void shutdown() {
+ }
+ }
+}
