fix(state): persist ReliableDispatcher.pending to DeliveryStateStore (issue #5301 Sub-PR B, fixes #5294 #5295) - #5311
Merged
Conversation
…(issue apache#5301 Sub-PR B, fixes apache#5294 apache#5295) 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. * new DeliveryStateStore interface in ...runtime.state: - put/remove/get/iterate/count/flush/close - Record carries (deliveryId, topic, partition, offset, clientId, attempt, nextAttemptAtMs, encodedEvent) - channel and mqAckCallback are runtime references that do NOT survive restart - toDelivery() rebuilds a live Delivery for tick/iterate use * InMemoryDeliveryStateStore (test contract baseline) * RocksDBDeliveryStateStore (production; key=deliveryId, value=ASCII line with base64-wrapped event bytes; mirrors RocksDBOffsetStore's pattern) * ReliableDispatcher now depends on a DeliveryStateStore (constructor parameter, default InMemory). The previous in-memory ConcurrentHashMap is gone; tick() iterates the store, ack() removes via the store, deliver() persists via the store. * New recover(): on startup, walk the store, write the stored offset as if the client had ACKed, advance the MQ physical cursor, 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. * package-info updated: DeliveryStateStore is now a real link, not a placeholder. Tests: * DeliveryStateStoreTest: in-memory + RocksDB contract harness (put/remove/get/ count/iterate; rocksDbPersistsAcrossClose) * DeliveryRecoveryTest: crash mid-delivery, fresh dispatcher, recover() retires all in-flight records and advances the OffsetStore without re-running the channel; idempotency; empty store no-op; nextAttemptAtMs round-trip Closes apache#5294 (unacked broadcast deliveries lost during cursor recovery) and apache#5295 (RocketMQ 5 POP broker ACK not gated on distribution completion).
…#5301) - RocksDBDeliveryStateStore: reorder imports to java.* before org.* (ImportOrder) - DeliveryStateStore: close cross-line {@link} and {@code} inline tags in Javadoc - package-info: break line 31 to under 150 chars (LineLength) - DeliveryStateStoreTest: rename local var aUpdated to aupdated (LocalVariableName) - DeliveryRecoveryTest: add EmptyLineSeparator between TestChannel field and @OverRide method
apache#5301 Sub-PR B) - After deliveryId was prefixed to the value, decode's offset/clientId/attempt still read the pre-shift ranges, so attempt = parseInt(clientId field) threw NumberFormatException and every RocksDB round-trip (contract + persist-across-close) failed. partition was also off by one. - Rewrite decode so each field reads its correct (pK+1, pK+1) range: deliveryId [0,p1), topic [p1+1,p2), partition [p2+1,p3), offset [p3+1,p4), clientId [p4+1,p5), attempt [p5+1,p6), nextAttemptAtMs [p6+1,p7), event [p7+1,end). - recover() offset advance is monotonic (writeOffset = max), so recovery is order-independent and DeliveryRecoveryTest's 101L assertion holds.
This was referenced Aug 28, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Sub-PR B: DeliveryStateStore (issue #5301 §DeliveryStateStore)
Makes in-flight delivery state survive a hard JVM restart. Every delivery is now persisted to a
DeliveryStateStoreon deliver/ack/nack/tick, and on startup the dispatcher'srecover()re-ACKs each persisted record against theOffsetStoreso no client is left without a progress advance and no delivery becomes an orphan.What's in this PR
DeliveryStateStore— new interface inorg.apache.eventmesh.runtime.state:put(Record) / remove(String) / get(String) / iterate(Consumer<Record>) / count() / flush() / close()Recordcarries(deliveryId, topic, partition, offset, clientId, attempt, nextAttemptAtMs, encodedEvent)— channel and mqAckCallback are runtime references that do NOT survive restartRecord.toDelivery()rebuilds a liveDeliveryfor tick / iterate useInMemoryDeliveryStateStore— test contract baseline (ConcurrentHashMap-backed)RocksDBDeliveryStateStore— production; key=deliveryId, value=ASCII line with base64-wrapped event bytes. Mirrors theRocksDBOffsetStorepattern (no extra dependency for the same field count).ReliableDispatchernow depends on aDeliveryStateStore(constructor parameter, defaultInMemory). The previous in-memoryConcurrentHashMapis gone:tick()iterates the storeack()/nack()remove / put via the storedeliver()persists via the storerecover()— 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.UniIngressServicewiresdispatcher.recover()into its constructor so a fresh JVM picks up unacked deliveries from a prior process without an explicit boot step.package-infoupdated:DeliveryStateStoreis now a real Javadoc link, not a placeholder.Tests
DeliveryStateStoreTest— in-memory + RocksDB contract harness (put/remove/get/count/iterate, includingrocksDbPersistsAcrossClose)DeliveryRecoveryTest— fault injection: crash mid-delivery, fresh dispatcher on the same store,recover()retires all in-flight records and advances theOffsetStorewithout re-running the channel; idempotency; empty-store no-op;nextAttemptAtMsround-tripCloses
Acceptance criteria touched
DeliveryStateStorerecovers across a hard restart — no orphaned in-flight deliveries ✓deliveryIdis now persisted viaDeliveryStateStore, 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)