qqeasonchen opened a new pull request, #5388:
URL: https://github.com/apache/eventmesh/pull/5388
fix(#5357): use immutable frame for MQ offset/partition stamps
## Summary
#5357 turned `EventMeshFrame` into an immutable value object whose
`attributes()` returns an unmodifiable map. All three storage plugins
were left mutating the frame in place to stamp the MQ physical
offset/partition after decoding each polled event. The
`UnsupportedOperationException` was swallowed by the surrounding
`catch (Exception decodeEx)`, which then retried the legacy
CloudEvents-JSON fallback against a binary frame body. Every polled
event was silently dropped.
## Changes
| File | Change |
|---|---|
|
`eventmesh-storage-plugin/eventmesh-storage-kafka/.../KafkaMeshStoragePlugin.java`
| `stampMqOffset` returns a new `EventMeshFrame` via `withAttribute`; both
decode + CE-JSON fallback paths keep the return value. |
|
`eventmesh-storage-plugin/eventmesh-storage-rocketmq/.../RocketMQRemotingStoragePlugin.java`
| New `stampMqOffset` helper (same shape as Kafka); replaces two in-place
`attributes().put` calls in the pull loop. |
|
`eventmesh-storage-plugin/eventmesh-storage-rocketmq5/.../RocketMQ5RemotingStoragePlugin.java`
| POP-mode `popCk` stamp via `frame.withAttribute(POP_CK_KEY, popCk)`. |
| `eventmesh-runtime/.../it/RealBrokerIntegrationTest.java` |
`maxAckedOffset()` walks every partition of the client — the broker picks the
queue, so partition 0 alone flaked 3 runs in 4 against a RocketMQ default
topic. |
|
`eventmesh-storage-plugin/eventmesh-storage-rocketmq/.../RocketMQRemotingStoragePlugin.java`
| `NettyClientConfig.connectTimeoutMillis` 2000 → 8000 to match the RocketMQ
5.x plugin (cold container / overlay network on first connect). |
## Impact
Single root cause produced four previously intermittent-looking test
failures. All trace back to the swallowed `UnsupportedOperationException`:
| Symptom | Root cause |
|---|---|
| `KafkaClientE2EIntegrationTest` "event delivered within 40s" |
`KafkaMeshStoragePlugin.stampMqOffset` mutation |
| `StreamingSdkE2ETest` token order shuffled |
`RocketMQ5RemotingStoragePlugin` POP `popCk` mutation on lite topics |
| `StreamingSdkE2ETest` "SSE stream closed by server" | same — broken
framing surfaces as disconnect |
| `RealBrokerIntegrationTest` offset assertion flaky | separate bug —
partition 0 hard-code vs broker queue selection |
## Verification
- Full `./gradlew check -x spotlessJava -x
:eventmesh-architecture-guard:test` (mirrors `.github/workflows/ci.yml`
excluding the spotless-JDK21 known incompatibility) — BUILD SUCCESSFUL, 291
tasks, all `test` + `checkstyleMain` + `spotbugsMain` + jacoco across
`eventmesh-runtime`, the three storage plugins, and all transitive modules
green.
- End-to-end suite against containerized brokers (docker compose in
`e2e-infra/`):
- `:eventmesh-runtime:e2eTest5x --tests "*StreamingSdkE2ETest*"` — 5
passing, 1 skipped (requires real LLM API key).
- `:eventmesh-runtime:e2eTest5x --tests "*RocketMQ5*"` — passing.
- `:eventmesh-runtime:e2eTest4 --tests "*RocketMQ4*"` — passing.
- `:eventmesh-runtime:e2eTest5x --tests "*KafkaClient*"` — passing.
- `:eventmesh-runtime:e2eTest4 --tests "*RealBrokerIntegrationTest*"` —
passing.
## Reproduction (against any local RocketMQ 5.x)
```sh
docker compose -f e2e-infra/docker-compose.rmq5.yml up -d --wait
./gradlew :eventmesh-runtime:e2eTest5x \
--tests "org.apache.eventmesh.runtime.it.KafkaClientE2EIntegrationTest" \
-Dit.storage=kafka -Dit.kafka.bootstrap=127.0.0.1:9092 \
-Dit.kafka.user=eventmesh -Dit.kafka.password=eventmesh-pass
```
The Kafka test was the cleanest 100 % reproducer.
## Follow-ups (not in this PR)
- `MultiInstanceRocketMqIntegrationTest` first-publish rejection under
two-runtime + Nacos — separate investigation.
- `spotlessJava` task broken under JDK 21 (google-java-format
incompatibility). CI excludes it via `-x spotlessJava`; should be
filed separately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
