qqeasonchen opened a new pull request, #5311:
URL: https://github.com/apache/eventmesh/pull/5311
## Sub-PR B: DeliveryStateStore (issue #5301 §DeliveryStateStore)
Makes in-flight delivery state survive a hard JVM restart. Every delivery is
now persisted to a `DeliveryStateStore` on deliver/ack/nack/tick, and on
startup the dispatcher's `recover()` re-ACKs each persisted record against the
`OffsetStore` so no client is left without a progress advance and no delivery
becomes an orphan.
### What's in this PR
* **`DeliveryStateStore`** — new interface in
`org.apache.eventmesh.runtime.state`:
- `put(Record) / remove(String) / get(String) / iterate(Consumer<Record>)
/ count() / flush() / close()`
- `Record` carries `(deliveryId, topic, partition, offset, clientId,
attempt, nextAttemptAtMs, encodedEvent)` — channel and mqAckCallback are
runtime references that do NOT survive restart
- `Record.toDelivery()` rebuilds a live `Delivery` for tick / iterate use
* **`InMemoryDeliveryStateStore`** — test contract baseline
(ConcurrentHashMap-backed)
* **`RocksDBDeliveryStateStore`** — production; key=deliveryId, value=ASCII
line with base64-wrapped event bytes. Mirrors the `RocksDBOffsetStore` pattern
(no extra dependency for the same field count).
* **`ReliableDispatcher`** now depends on a `DeliveryStateStore`
(constructor parameter, default `InMemory`). The previous in-memory
`ConcurrentHashMap` is gone:
- `tick()` iterates the store
- `ack()` / `nack()` remove / put via the store
- `deliver()` persists via the store
- Retry-vs-ack race is guarded at the iterate-then-act layer (snapshot
semantics + re-read)
* **`recover()`** — new method: on startup, walk the store, write the stored
offset as if the client had ACKed, advance the MQ physical cursor
(`emmqoffset`/`emmqpartition`), remove the record. **Never re-runs the
channel** — the broker has already either redelivered (Kafka / RocketMQ 4.x
PULL) or considered the message gone (RocketMQ 5.x POP). Idempotent.
* **`UniIngressService`** wires `dispatcher.recover()` into its constructor
so a fresh JVM picks up unacked deliveries from a prior process without an
explicit boot step.
* **`package-info`** updated: `DeliveryStateStore` is now a real Javadoc
link, not a placeholder.
### Tests
* **`DeliveryStateStoreTest`** — in-memory + RocksDB contract harness
(put/remove/get/count/iterate, including `rocksDbPersistsAcrossClose`)
* **`DeliveryRecoveryTest`** — fault injection: crash mid-delivery, fresh
dispatcher on the same store, `recover()` retires all in-flight records and
advances the `OffsetStore` without re-running the channel; idempotency;
empty-store no-op; `nextAttemptAtMs` round-trip
### Closes
* **#5294** (unacknowledged broadcast deliveries lost during cursor recovery)
* **#5295** (RocketMQ 5 POP broker ACK not gated on distribution completion)
### Acceptance criteria touched
* `DeliveryStateStore` recovers across a hard restart — no orphaned
in-flight deliveries ✓
* ACK validation binds to delivery ownership — `deliveryId` is now persisted
via `DeliveryStateStore`, so a stale ACK cannot match a fresh delivery after
restart ✓
### Sub-PR ordering
A (PR #5310 ✓ merged) → **B (this PR)** → C (DeadLetterStore + TaskStore
Meta backends) → D (cross-store fault-injection)
--
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]
