[Bug] Do not retire deliveries when durable offset persistence fails

[qqeasonchen]

Execution meta — see plan: #5296

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

Problem

RocksDBOffsetStore.writeOffset() catches persistence errors and only increments a failure counter. ReliableDispatcher.ack() does not receive a write result, so it removes the pending delivery and can execute a RocketMQ 5 broker ACK even if the subscriber offset or physical cursor was not durably persisted.

Relevant code:

• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/offset/RocksDBOffsetStore.java: writeOffset()
• eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/delivery/ReliableDispatcher.java: ack(String)

Reproduction

1. Use an OffsetStore test double that fails writeOffset.
2. Deliver a message with a deferred broker ACK callback.
3. ACK the delivery.
4. Observe that the dispatcher retires the delivery and may invoke broker ACK despite offset persistence failure.
5. Restart from a pull cursor that is ahead of the durable ACK state.

Expected behavior

An ACK must not be considered successful until required durable state is written. If offset/delivery-state persistence fails, EventMesh must not execute source broker ACK and must preserve a retryable delivery state.

Proposed direction

• Change the offset persistence contract to return a result or throw a typed exception.
• Make ReliableDispatcher.ack() transactional in intent: validate → persist completion/offset → broker ACK → retire delivery.
• On persistence failure, retain or requeue the delivery and return a retryable error.
• Add health/metrics for durable ACK failures.

Acceptance criteria

• A failed offset write prevents broker ACK callback execution.
• A failed offset write does not permanently retire the delivery.
• The ACK endpoint reports failure/retryability to the client.
• Restart after an offset-write failure does not skip the message.
• Metrics expose durable ACK/offset write failures.
• Tests cover Kafka/RocketMQ 4 cursor recovery and RocketMQ 5 deferred broker ACK behavior.

Suggested labels

bug, reliability, storage, priority:p0

[qqeasonchen]

Fixed in #5307: OffsetStore.writeOffset now returns boolean. ReliableDispatcher.ack() only retires a delivery after the durable offset write succeeds; on failure the delivery stays in flight (backoff + redelivery) instead of being silently dropped.