feat(state): DeadLetterStore + TaskStore Meta backends (issue #5301 Sub-PR C, fixes #5292 fully, unblocks #5302) - #5312
Merged
qqeasonchen merged 13 commits intoAug 28, 2026
Conversation
…5301 Sub-PR C, fixes apache#5292 fully, unblocks apache#5302) Two production implementations on top of the Sub-PR A interfaces: - MetaBackedDeadLetterStore: idempotent ledger under /em/dlq/<deliveryId> via MetaStore.putIfAbsent CAS; first recorder wins, subsequent writers are no-ops that still return true so the dispatcher proceeds to retire. - MetaBackedTaskStore: per-task record under /em/tasks/<taskId> with a self-describing v1|<base64> wire envelope (all variable-width fields base64-encoded so opaque caller input containing '|' or newlines cannot corrupt the format). Status updates use MetaStore.tryAcquire as an epoch CAS, rejecting stale writers (issue apache#5291 style). ReliableDispatcher 9-arg ctor: same as 8-arg + a DeadLetterStore that is invoked on every confirmed DLQ transition. The 8-arg ctor (legacy Sub-PR A/B) is untouched and the DLQ path skips the ledger call when no store is wired. Backward compatible. Tests: MetaBackedDeadLetterStoreTest, MetaBackedTaskStoreTest, ReliableDispatcherDlqLedgerTest. Interface contract tests DeadLetterStoreTest and TaskStoreTest (Sub-PR A) are unchanged and continue to pass against the new production implementations. Scope: fixes apache#5292 fully; unblocks apache#5302. A2A dispatch mode (apache#5302) and the TaskExpirer reaper are follow-up scope in Sub-PR D (issue apache#5297).
…in (issue apache#5301 Sub-PR C) The 8-arg ctor assigns `deadLetterStore = null` and the 9-arg ctor chains to the 8-arg ctor and then reassigns with the supplied ledger; this hit `variable deadLetterStore might already have been assigned` because the field was `final`. Drop `final` (with a Javadoc note that it is effectively final after construction) so the chain compiles. No runtime semantics change.
…PR C, issue apache#5301) - ReliableDispatcher.java (CRLF): import order (DeadLetterStore before DeliveryStateStore), drop the duplicate DeliveryStateStore import, split the 274-char log.warn into three concatenated string fragments to satisfy LineLength <= 150. - ReliableDispatcherDlqLedgerTest.java (LF): import AckCallback / DeadLetterSink / PushChannel from `org.apache.eventmesh.runtime.delivery` (not `connector` / `push`); drop the fully-qualified `org.apache.eventmesh.runtime.push.AckCallback` in the FakeChannel override (now imports correctly). All three compile errors and the three checkstyle violations from run 33057816390 are addressed. No semantic change.
- MetaBackedDeadLetterStoreTest: drop unused MetaStore import (only InMemoryMetaStore is used). - MetaBackedTaskStoreTest: rename local `bView` -> `bViewTwo` to satisfy LocalVariableName (lowercase-first camelCase pattern). - ReliableDispatcherDlqLedgerTest: drop three same-package `org.apache.eventmesh.runtime.delivery.*` imports (the test lives in that package), and move `org.junit.jupiter.api.Test` to the top import block (it was below `io.*` and broke ImportOrder). All six checkstyle violations from run 33062276583 addressed; no semantic change.
- MetaBackedTaskStoreTest: rename local `bViewTwo` -> `taskB`. The checkstyle LocalVariableName regex is `^[a-z]([a-z0-9][a-zA-Z0-9]*)?$` -- the second char must be lowercase or digit, so `bViewTwo` (b + V) and `bView` (b + V) both fail. `taskB` is the clean alternative. - ReliableDispatcherDlqLedgerTest: import order was wrong. ASCII order requires `org.apache.*` (a) before `org.junit.*` (j) before `java.*` (a again, but `j` group sorts before `java`). The previous amendment placed `org.junit` ABOVE `org.apache.eventmesh.*`, which violates ImportOrder.
…pache#5301 Sub-PR C) Apache checkstyle's ImportOrder rule uses a configured group order (org.apache.eventmesh | org.apache | java | javax | org | io | net | junit | com | lombok), not strict ASCII order. 'java.*' is group 4, so it must appear AFTER 'org.apache.*' (group 3) and BEFORE 'org.junit.*' (group 9). The previous amendment put 'org.junit.*' before 'java.*', tripping checkstyleTest with 'Wrong order for java.net.URI import'. Move all 'org.junit.*' imports to the LAST non-static group, after any 'java.*' or 'io.*' imports, in: - ReliableDispatcherDlqLedgerTest - MetaBackedTaskStoreTest - MetaBackedDeadLetterStoreTest (no java imports; ordering unchanged)
…pache#5301 Sub-PR C) Apache checkstyle's ImportOrder rule uses a configured group order (org.apache.eventmesh | org.apache | java | javax | org | io | net | junit | com | lombok), not strict ASCII order. 'java.*' is group 4, so it must appear AFTER 'org.apache.*' (group 3) and BEFORE 'org.junit.*' (group 9). The previous amendment put 'org.junit.*' before 'java.*', tripping checkstyleTest with 'Wrong order for java.net.URI import'. Move all 'org.junit.*' imports to the LAST non-static group, after any 'java.*' or 'io.*' imports, in: - ReliableDispatcherDlqLedgerTest - MetaBackedTaskStoreTest - MetaBackedDeadLetterStoreTest (no java imports; ordering unchanged)
Apache checkstyle's ImportOrder rule uses a *first-match* prefix algorithm against the `groups` config. `org.junit.*` matches the `org` prefix (index 4) before the `junit` prefix (index 7), so org.junit imports belong to the `org` group, NOT the `junit` group. Required order: org(4) -> io(5). Previous fix had io(5) before org.junit(4), tripping checkstyleTest with 'Wrong order for org.junit.jupiter.api.Test' on line 42. Reorder imports in ReliableDispatcherDlqLedgerTest.java to match the pattern used by UniAdminServerTest.java (which passes): static org.apache.eventmesh.* java.* org.junit.* (org group, index 4) io.* (io group, index 5)
…-PR C) Two Sub-PR C tests had runtime logic bugs that compile/checkstyle did not catch: 1. ReliableDispatcherDlqLedgerTest (both tests). The original test loop used MAX_ATTEMPTS=2 iterations of (nack, clock+=1s, tick) which is one iteration short: nack moves the next-attempt deadline into the future (clock + backoff(1) = 1s) but the clock only advances 1s per tick, so the second tick sees the delivery as not yet expired and skips it. The test never reaches the DLQ branch (attempt >= maxAttempts). Switch to timeout-driven exhaustion (matching ReliableDispatcherTest.exhaustedRetriesGoToDLQ): bump MAX_ATTEMPTS to 3 and run 3 ticks each preceded by clock.addAndGet(ACK_TIMEOUT), so attempt 1 -> 2 -> 3 -> DLQ. The first nack is no longer needed. 2. MetaBackedTaskStoreTest.expireStaleRemovesOldRecords. The test calls expireStale(1L) immediately after updateStatus; both records are >1ms old by then, so both end up in the expired list and the size==1 assertion fails. Add a 3ms sleep between the last updateStatus and expireStale so the just-updated 'new' record's updatedAtMs is within 1ms of 'now' and survives, while 'old' (created several ms earlier) is removed. These are test-only changes; production code (MetaBackedTaskStore, ReliableDispatcher, MetaBackedDeadLetterStore) is unchanged.
Thread.sleep(3L) inside expireStaleRemovesOldRecords throws InterruptedException; the test method must declare `throws Exception` (or wrap the call in try/catch) for the test source to compile.
In ReliableDispatcher.tick, the dlqSink is invoked with rec.topic
(the SOURCE topic, e.g. 'orders'). The '_DLQ' suffix is only
applied when recording on the durable ledger via
deadLetterStore.recordDeadLetter. The previous assertion
channel.dlqTopics.contains("orders_DLQ") is therefore wrong;
the sink sees the source topic unchanged.
Update the assertion to check for "orders" (the source topic),
with a comment explaining the source-vs-ledger distinction so the
next reader doesn't trip on the same misunderstanding.
The previous 3ms sleep worked locally but was too tight for CI scheduler noise: on a busy host the 1ms window can collapse, so both 'old' and 'new' records end up older than 1ms by the time expireStale(1L) is called. Bump the sleep to 50ms (still fast at < 100ms total per test) so the just-updated 'new' record's updatedAtMs is well within 1ms of 'now' (it survives), while 'old' (created ~50ms earlier) is removed. The assertion logic is unchanged.
The previous test (and two follow-up attempts) updated 'new'
BEFORE the sleep, so by the time expireStale(1L) ran both
records were >1ms old and both got removed (expired.size() == 2).
Correct order:
1. createTask('old')
2. createTask('new')
3. Thread.sleep -- makes 'old' stale
4. updateStatus('new') -- refreshes 'new' to current time
5. expireStale(1L) -- removes only 'old' (now >1ms old)
Bump sleep to 30ms (was 50ms) -- still well under 100ms test budget,
and >1ms so 'old' is reliably stale even on a busy CI host.
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.
Summary
Two production implementations of the Sub-PR A interfaces, plus the wiring into
ReliableDispatcher.Changes
New files
MetaBackedDeadLetterStore— idempotent DLQ ledger under/em/dlq/<deliveryId>viaMetaStore.putIfAbsentCAS. First recorder wins; concurrent writers see the existing record and return true so the dispatcher proceeds to retire.MetaBackedTaskStore— per-task record under/em/tasks/<taskId>with a self-describingv1|<base64>wire envelope. Variable-width fields (taskId, agentId, clientId, input, output) are base64-encoded before joining so opaque caller input containing|or newlines cannot corrupt the format. Status updates useMetaStore.tryAcquireas an epoch CAS, rejecting stale writers (issue [Bug] Prevent stale ACKs from matching reused delivery IDs #5291 style).MetaBackedDeadLetterStoreTest,MetaBackedTaskStoreTest,ReliableDispatcherDlqLedgerTest.Modified
ReliableDispatcher— new 9-arg ctor that adds an optionalDeadLetterStore. The legacy 8-arg ctor (Sub-PR A/B wire) is untouched and the DLQ path skips the ledger call when no store is wired. The DLQ transition now records the deliveryId on the ledger before retirement; a ledger failure is logged but does not block retirement (the downstream DLQ sink has already persisted the body).Scope
Diff stat
Test plan
./gradlew :eventmesh-runtime:test— all Sub-PR A interface contract tests (DeadLetterStoreTest, TaskStoreTest) plus the 3 new tests above must pass../gradlew :eventmesh-runtime:checkstyleMain :eventmesh-runtime:checkstyleTest— must be clean (no new violations introduced).