test(e2e): real-broker coverage for BatchPublish, Unsubscribe, RateLimit, RequestReply - #5389

qqeasonchen · 2026-09-09T15:46:10Z

What this PR adds

Four feature-level e2e tests under :eventmesh-runtime:test-e2e, each driving a real RocketMQ broker end-to-end. These were previously covered only via UniIngressService + in-memory MeshStoragePlugin stubs, which means regressions in the broker-boundary code path (EventMeshFrame serialization, subscription lifecycle crossing the storage layer, partition routing in lite-pull consumer) had no failing signal.

Test	What it pins down
`BatchPublishOverBrokerTest`	A 10-event `publishBatch` round-trips over real `LitePullConsumer` — every event ID arrives.
`UnsubscribeOverBrokerTest`	After `unsubscribe(topicA)`, topic A events no longer reach the subscriber while topic B continues — this is the case the in-process stub cannot catch, since dispatcher bookkeeping ≠ real-broker delivery routing.
`RateLimitOverBrokerTest`	A 2-token bucket with 0 refill: 5 publishes produce 2 successes + 3 `RateLimitedException` rejects, AND the `rateLimited` counter ticks up by 3. The counter assertion is what catches "the limiter silently drops instead of rejecting".
`RequestReplyOverBrokerTest`	A `/req` round-trip completes end-to-end over the broker, proving the request survives subscription rebalance before the responder's reply comes back.

Supporting changes

FeatureBrokerHarness: shared bootstrap that picks rocketmq (4.9 plugin) or rocketmq5 based on -Dit.feature.storage, looks up the storage plugin via SPI (the same path production uses), boots UniRuntime, and tears it down. Skips cleanly when the plugin class is not present, so CI without a broker stays green.
e2eFeature gradle task: runs all four (or a selected one with -Pfeature.test=FQN). Reuses the existing rocketmq4TestRuntimeClasspath so storage-rocketmq5 is excluded and the 4.9 plugin wins — same contract as e2eTest4.
BrokerDiscoverer: serialize admin RPC across the test JVM with a static lock. Without it, four tests creating topics back-to-back thrash the broker's single-shard topic-create / heartbeat thread and the namesrv route doesn't propagate in time ("no broker for topic queue 0").

Verified

All four tests pass in isolation against RocketMQ 4.9.5 over real UniRuntime on macOS / aarch64.
./gradlew check -x spotlessJava -x :eventmesh-architecture-guard:test remains green across storage-plugin (rocketmq, rocketmq5, kafka) and runtime modules.
The e2eFeature task is gated on -Dit.feature.storage; CI without a broker stays green. CI does not run it (opt-in gradle task outside the default build chain), so this PR adds no CI surface.

Known broker artifact

Running all four tests in a single gradle :eventmesh-runtime:e2eFeature invocation on a single-broker cluster is flaky under topic-creation pressure — the broker's topic-create / heartbeat thread becomes the bottleneck and routes don't propagate in time. Per-test (-Pfeature.test=FQN) runs are deterministic. Prefer -Pfeature.test=… in CI scripts that opt into this gate.

Builds on #5388 (eventmesh-frame immutability, which made these tests viable again on RocketMQ5).

…stReply The previous e2e suite covered the publish/subscribe/streaming/legacy-tcp production paths over a real broker (rocketmq4 / rocketmq5 / kafka + nacos for legacy cluster). Half a dozen equally important feature behaviours — batch publish, topic unsubscribe, rate-limit token bucket, request-reply routing — were only exercised over UniIngressService + an in-memory MeshStoragePlugin stub. Anything that depended on the broker boundary (frame serialization, topic partition routing, subscription lifecycle crossing the storage layer) could regress without any test failing. Add four focused tests under the same broker-IT discipline (skipped when no broker is configured, gated on a system property): - BatchPublishOverBrokerTest — publishes a 10-event batch, verifies every event arrives over the real LitePullConsumer path. - UnsubscribeOverBrokerTest — subscribes to two topics, unsubscribes one, confirms the dropped topic stops receiving while the survivor keeps working (this is the failure mode that caused the in-process stub to be insufficient: dispatcher-level topic bookkeeping != broker delivery routing). - RateLimitOverBrokerTest — sets a 2-token bucket with 0 refill, publishes 5 events, asserts 2 succeed + 3 fail with RateLimitedException AND the rateLimited counter ticks up by 3. The counter assertion is what catches "the limiter silently drops instead of rejecting" — exactly the kind of behaviour the in-process test papers over. - RequestReplyOverBrokerTest — opens a /req round-trip end-to-end, proving the request survives broker round-trip + subscription rebalance before the responder's reply comes back. Supporting changes: - FeatureBrokerHarness: shared bootstrap. Picks rocketmq or rocketmq5 backend by it.feature.storage, looks up the storage plugin via SPI (the same path the real runtime uses), boots UniRuntime, and closes it on tearDown. Skips cleanly when the plugin isn't on the classpath so the build stays green without a broker. - e2eFeature gradle task: selects tests by -Pfeature.test=FQN (without the property, all four run). Uses the existing rocketmq4TestRuntimeClasspath so storage-rocketmq5 is excluded and the 4.9 plugin wins, matching the e2eTest4 contract. - BrokerDiscoverer: serialize admin RPC across the test JVM. Without the lock, four tests creating topics back-to-back thrash the broker's single-shard topic-create / heartbeat thread and the route doesn't propagate in time (the "no broker for topic queue 0" failure seen during batch runs). Verified: - All four tests pass in isolation against a real RocketMQ 4.9.5 broker over a uni-runtime in UniRuntime-mode on macOS/aarch64. - Storage plugins (rocketmq, rocketmq5, kafka) and runtime unit tests remain green via ./gradlew check -x spotlessJava -x :eventmesh-architecture-guard:test. - The e2eFeature task is gated on -Dit.feature.storage so CI without a broker stays green. CI does not run it (it's an opt-in gradle task outside the default build chain). The "all four in one gradle invocation" run is flaky on a single-shard broker under topic-creation pressure; that's a broker-capacity limit, not a test correctness bug. Running with -Pfeature.test=FQN is deterministic.

Bring the single source of truth up to date with what actually landed: - SSE / WS push: unified ACK-tracked redelivery + DLQ now shared with long-polling via ReliableDispatcher; real-broker e2e suite in #5389 - Connector Runtime: all 23 plugins fully implemented since #5394 (the "rest are templates" note is stale); unit tests still cover only file/kafka/pulsar/rocketmq - A2A: task reaper + Meta-backed agent cards landed in #5346, quota classification in #5373; Beta promotion gated on Testcontainers E2E (#5340) instead of "pending" items - add the memory storage backend row (default, zero-dependency dev/CI; #5400) and mention it in the features list, both languages

…in, A2A to 10108, 10205 gRPC reserved 8080/8081 collide with the most common dev-service ports (Tomcat, Jenkins, Spring Boot apps, proxies). Move back onto the project's own 1.x default port block, which keeps long-time users on familiar numbers:/n - traffic HTTP 8080 -> 10105 (all app protocols multiplex on it, incl. the legacy /eventmesh/* bridge) - admin HTTP 8081 -> 10106 - WebSocket 8082 -> 10107 (opt-in, unchanged default -1) - A2A Gateway 10105 -> 10108 (frees 10105 for the CORE runtime; the experimental gateway takes the adjacent 10108) - 10205 documented as RESERVED for a future gRPC protocol port (the 1.x default): the uni runtime does not ship a gRPC server yet, and the port must not be casually taken so a later gRPC addition lands on the historical number Touched: EventMeshApplication defaults, bin/start.sh, docker/Dockerfile(+.connector), K8s manifests + README, SDK CloudEventsClient default URL + javadocs, A2AClient / ExampleConstants / A2AGatewayDemo, e2e test literals, docs (quickstart, configuration, deployment, a2a) and eventmesh.properties comments. Configuration keys are unchanged - only defaults move; existing deployments pinning the old ports keep working. Also point image-smoke-test.yml at the new ports throughout (host = container, 10105/10106 - no offset mapping left). README / README.zh-CN quick-start port mentions updated too. Feature docs swept too (client-java, pubsub, streaming, introduction, index, admin-api, http-api, observability, connectors overview) - connector-local webhook ports are untouched. Fix a port-sweep typo (10100 -> 10105) in the documentation map. README / README.zh-CN capability-status refreshed against the current tree:/n- SSE/WS push: ACK-tracked redelivery + DLQ shared with long-polling (ReliableDispatcher); e2e-real-broker suite in #5389 - Connector Runtime: all 23 plugins fully implemented since #5394 (no template stubs left); 4 of 23 still carry the only unit tests - A2A Gateway: reaper + Meta-backed agent cards landed #5346, quota #5373; Beta promotion gated on Testcontainers E2E (#5340) - new Memory storage row (Beta, zero-dependency dev/CI default, #5400)

qqeasonchen added the improvement Improve the mechanism or performance label Sep 9, 2026

qqeasonchen merged commit 815eecc into apache:develop Sep 9, 2026
1 check passed

Navigation Menu

Repository navigation

test(e2e): real-broker coverage for BatchPublish, Unsubscribe, RateLimit, RequestReply - #5389