[I] Storage plugins silently drop every polled event after #5357 (eventmesh)

["qqeasonchen (via GitHub)" <gi...@apache.org>]

qqeasonchen opened a new issue, #5387:
URL: https://github.com/apache/eventmesh/issues/5387

   # Storage plugins silently drop every polled event after #5357
   
   PR #5357 turned `EventMeshFrame` into an immutable value object whose
   `attributes()` returns an unmodifiable map. All three storage plugins
   were left calling `attributes().put(...)` to stamp the MQ physical
   offset/partition after decoding each polled event:
   
   ```
   EventMeshFrame frame = EventMeshFrame.decode(msg.getBody());
   frame.attributes().put("emmqoffset", Long.toString(msg.getQueueOffset())); // throws UOE
   frame.attributes().put("emmqpartition", Integer.toString(msg.getQueueId()));
   ```
   
   That `UnsupportedOperationException` is swallowed by the surrounding
   `catch (Exception decodeEx)` block, which then retries the legacy
   CloudEvents-JSON fallback against a binary frame body. The fallback
   decodes nothing and the polled event is silently dropped — subscribers
   starve and the OffsetStore never sees the cursor advance.
   
   ## Impact
   
   - **Kafka**: 100 % reproducer. Every polled event dies; the
     `KafkaClientE2EIntegrationTest` "event delivered within 40s" assertion
     fails on the very first message.
   - **RocketMQ 5.x**: only fires when `popCk != null` (lite topics that
     exercise POP). `StreamingSdkE2ETest` over rmq5 lite topics hits
     this — token-order and SSE-closed failures all traced back to the
     same swallowed exception.
   - **RocketMQ 4.x**: same as Kafka. Was masked earlier in the e2e
     suite by other failures in this commit's parent.
   
   ## Reproduction
   
   ```
   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
   ```
   
   ## Suggested fix
   
   Derive a new frame via `withAttribute` instead of mutating in place.
   Callers must keep the return value.
   
   ```java
   private static EventMeshFrame stampMqOffset(EventMeshFrame frame, int partition, long offset) {
       return frame.withAttribute("emmqoffset", Long.toString(offset))
           .withAttribute("emmqpartition", Integer.toString(partition));
   }
   ```
   
   PR with the fix: linked below.


-- 
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.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: [I] Storage plugins silently drop every polled event after #5357 (eventmesh)

["qqeasonchen (via GitHub)" <gi...@apache.org>]

qqeasonchen closed issue #5387: Storage plugins silently drop every polled event after #5357
URL: https://github.com/apache/eventmesh/issues/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