You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@eventmesh.apache.org by "qqeasonchen (via GitHub)" <gi...@apache.org> on 2026/09/09 15:09:39 UTC
[PR] fix(#5387): use immutable frame for MQ offset/partition stamps (eventmesh)
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: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org
Re: [PR] fix(#5387): use immutable frame for MQ offset/partition stamps (eventmesh)
Posted by "qqeasonchen (via GitHub)" <gi...@apache.org>.
qqeasonchen commented on PR #5388:
URL: https://github.com/apache/eventmesh/pull/5388#issuecomment-5604147111
Fixes #5387
--
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: issues-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: issues-help@eventmesh.apache.org
Re: [PR] fix(#5387): use immutable frame for MQ offset/partition stamps (eventmesh)
Posted by "qqeasonchen (via GitHub)" <gi...@apache.org>.
qqeasonchen merged PR #5388:
URL: https://github.com/apache/eventmesh/pull/5388
--
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: dev-unsubscribe@eventmesh.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: dev-help@eventmesh.apache.org