[Bug] Preserve unacknowledged broadcast deliveries during cursor recovery

[qqeasonchen]

Execution meta — see plan: #5296

• Phase: 2 — unified state control plane
• Severity: P0
• Depends on: [Architecture Review][P0] Unified state control-plane stores #5301
• Blocks: (none)

Problem

The Runtime records one physical MQ cursor per topic + partition under the reserved __mqcursor__ client ID whenever any delivery is ACKed. In broadcast or multicast delivery, the same physical message is delivered to multiple targets. A fast target can advance this shared cursor while another target still has an unacknowledged copy.

On restart, UniRuntime.alignPullOffsetsToAck() uses the shared cursor to rewind Kafka/RocketMQ 4 pull state. The model cannot represent per-subscriber completion obligations, so a slow target's unacknowledged copy can be skipped or replay behavior can become backend-dependent.

Relevant code:

• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java: pullAndDispatchPartition()
• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/delivery/ReliableDispatcher.java: ack(String)
• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/boot/UniRuntime.java: alignPullOffsetsToAck()

Reproduction

1. Use Kafka or RocketMQ 4 with a topic/partition containing one message.
2. Register two BROADCAST subscribers, A and B.
3. Deliver the message to both subscribers.
4. ACK from A only; leave B unacknowledged.
5. Crash/restart the Runtime before B ACKs.
6. Verify whether B receives the message again. The required result must not depend on which subscriber ACKed first.

Expected behavior

For broadcast/multicast, recovery must preserve every uncompleted delivery obligation. A physical cursor must not advance past a message until the configured completion rule for that distribution mode is satisfied.

Proposed direction

• Define recovery semantics independently for LOAD_BALANCE, BROADCAST, and MULTICAST.
• Track physical progress per subscription/group, or persist delivery obligations keyed by topic, physical partition, physical offset, subscription/client, and completion state.
• Derive restart rewind from the minimum incomplete physical position rather than a single global __mqcursor__ value.
• Document the duplicate-delivery bound for each mode.

Acceptance criteria

• A fast BROADCAST subscriber ACK cannot cause an unacknowledged peer to lose recovery eligibility.
• Equivalent tests exist for MULTICAST and LOAD_BALANCE semantics.
• Restart recovery is verified with Kafka and RocketMQ 4 or storage-contract stubs that model physical cursors.
• Cursor/obligation state is durable across a hard Runtime restart.
• Admin/metrics distinguish physical cursor state from per-subscriber completion state.

Suggested labels

bug, reliability, architecture, priority:p0