GeezerVison

…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.