feat(core) #197: carry the caller's MDC into the threads that run your function - #205
feat(core) #197: carry the caller's MDC into the threads that run your function#205astubbs wants to merge 6 commits into
Conversation
…r function A caller who had established diagnostic context - a trace_id, a request_id, a tenant - lost all of it the moment a record crossed into the worker pool, and again crossing into vert.x, Reactor or Mutiny. The logs their user function wrote could not be correlated back to the originating request. PC set only its own two keys (pcId, offset) and called neither MDC.getCopyOfContextMap() nor MDC.setContextMap() anywhere in the tree. The context is snapshotted on the thread that calls poll*() - the only moment it is reachable, since none of PC's threads exist yet and the SLF4J MDC is not inheritable - and re-established around the user function on the thread that actually runs it, then torn down again. The teardown is the half that matters. Worker threads are pooled: one that keeps the previous task's trace_id produces logs that are actively misleading, which is worse than no context at all. That leak already existed on master for the user function's own MDC.put calls - five records on one pooled thread carry four poisoned keys - and is fixed by the same scope. Precedence: PC's own pcId/offset are applied AFTER the caller's map at every call site, so a caller key of the same name cannot shadow them. PC's own log lines are read by those keys. On by default, with ParallelConsumerOptions.propagateMdc as the kill switch. Not propagating fails silently for everyone; propagating fails visibly, as one extra key in a log line, and can be switched off. Setting it false restores the old behaviour exactly, leak included, so it is a true revert rather than half of one. PC logs the captured keys - never the values - at INFO, so that a request-scoped value accidentally pinned for the life of the consumer is discoverable rather than silently wrong. Cost on the hot path is nil for anyone who never touches the MDC: getCopyOfContextMap() returns null, allocating nothing, for an empty context, and pcId is only ever set when setMyId() has been used. Deliberately not covered: Reactor's and Mutiny's own operator chains. We carry the context into the user's function and into PC's terminal signal handling; following it through the Publisher/Uni they return needs Reactor's own io.micrometer:context-propagation and is the user's call, not ours. Upstream-Issue: confluentinc#907 Forwarded: not-needed Applied-Upstream: no Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RtNUsxokE9g2pSEjBHZqNA
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Scanned FilesNone |
Claude is reviewing this PR
|
✅ Duplicate Code ReportTwo engines run in parallel for cross-validation. Each has its own thresholds tuned to its baseline - the real safety net is the per-engine "max increase vs base" check. ✅ PMD CPD
No new clones introduced by this PR. ✅ jscpd (language-agnostic)
|
…ntry Lifecycle transition per AGENTS.md - the manifest is the source of truth for the fork side, and nothing automated checks it, so an entry saying prs: [] with an open PR rots silently. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RtNUsxokE9g2pSEjBHZqNA
|
Claude finished @astubbs's task in 9m 8s —— View job Claude's review of #205Tasks
SummaryNo issues found. This is a careful, well-scoped change and the PR body's claims check out under independent verification:
One minor accuracy nit (non-blocking)The PR checklist says "13 new tests across 4 modules." Actual count is 17: |
✅ SpotBugs ReportNo bugs found (new bugs only — baseline from base branch excluded). |
… tests The file-similarity report flagged the three new engine tests as newly introduced duplication (Mutiny<->Reactor 69%, Mutiny<->vertx 55%, Reactor<->vertx 53%). The identical part - the caller key/value, the two observation queues, the MDC teardown, and the "context was visible" assertion - now lives once in MdcBoundaryProbe, in the core test-jar, the same route TestConventionRules and WireMockUtils already take. Used by composition, not a shared base class: each engine test already extends its own module's unit-test base, so inheritance is not available. The per-engine assertions deliberately stay in the tests, because they are what stops a test quietly degrading into a re-run of the core worker-pool case, and they genuinely differ: Reactor asserts its scheduler thread positively (boundedElastic), while vertx and Mutiny can only assert "not a pc- thread" since their executor is caller-supplied. Nothing was weakened - the probe's failure messages now also report the threads and values seen, which the previous assertions did not. Re-verified all three can still go red: temporarily overriding initAsyncConsumer with propagateMdc(false) fails each on its own "context must be visible on the <engine>" assertion with `values seen: [null, ...]`. Also: - the escape-hatch test's javadoc read as though propagation were off by default; say plainly that it is on - record the measured allocation cost in the inflight note: 0 bytes per batch when the caller never touches the MDC (measured with ThreadMXBean.getThreadAllocatedBytes, not assumed), ~424 bytes when they do, with the caveat that the zero is a logback MDCAdapter property - note that the startup INFO line is structurally once-per-instance, since supervisorLoop throws if poll*() is called twice Refs #197, #195, upstream confluentinc#907. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RtNUsxokE9g2pSEjBHZqNA
|
Responding to the review in #205 (comment) and to the two automated duplication reports. Fixed in 544347b. 1. "13 new tests" was wrong — you were right, it is 17. FixedCounted independently rather than taking the number on faith, and the reviewer's count is exactly right:
Confirmed by surefire, not just by grep — core reports 2. Startup INFO line — checked proactively, it is once per instanceWorth stating explicitly since #203 is specifically about log noise, and an INFO line on a hot path would be a genuine regression. It is not on one, and structurally cannot be: if (state != State.UNUSED) {
throw new IllegalStateException(msg("Invalid state - you cannot call the poll* or pollAndProduce* methods more than once ..."));
}so it runs once per consumer, at startup, before any PC thread exists. Not per poll, not per record. Now recorded in the inflight note so it stays that way. 3. Hot-path cost — measured rather than reasoned aboutThe PR claimed "nil for users who never touch the MDC" on the strength of an argument. Measured it instead, with
So the claim holds exactly: zero bytes, not merely "small". One honest caveat I have added to the PR body and the inflight note: that zero is a logback 4. File-similarity report — refactoredThe similarity check flagged the three new engine tests as newly introduced duplication (Mutiny↔Reactor 69.0%, Mutiny↔vertx 55.2%, Reactor↔vertx 53.1%). That is duplication this PR introduced, so it is in scope and is now removed rather than argued away. The identical part — caller key/value, the two observation queues, the MDC teardown, and the "context was visible" assertion — now lives once in What deliberately did not move: the per-engine assertions. Those are precisely what stops a test quietly degrading into a re-run of the core worker-pool case, and they genuinely differ — Reactor asserts its scheduler thread positively ( Nothing was weakened; the assertions got slightly stronger. The probe's failure messages now report the threads and the values actually seen, which the previous hand-rolled assertions did not. And all three were re-verified to still go red after the extraction — temporarily overriding A refactor that silently defused the tests would have been worse than the duplication. 5. Duplicate-code report, jscpd's "+1 new clone" — pushing back, no changejscpd flagged an 8-line clone at Worth noting the cross-check engine disagrees in the right direction: PMD CPD reports −1 clone and −28 duplicated lines for this PR, and overall duplication falls on both engines (CPD 2.51% → 2.43%, jscpd 3.26% → 3.20%). Both checks pass. 6. One thing for the maintainer, not for me —
|
|
Claude finished @astubbs's task in 8m 37s —— View job Claude's review of #205 (re-review after commit 544347b)Tasks
SummaryNo issues found. This is the follow-up commit (
Did not re-run the destructive "flip No other findings. The response to the first round (test-count fix, probe extraction, measured-cost table, once-per-instance clarification) addresses everything raised, and nothing new turned up on a fresh pass. |
|
Closing the loop on 544347b — all 25 checks green, and the re-review came back clean ("No issues found", having independently re-run all 17 tests and hand-traced the MDC precedence order). The probe extraction worked, measurablyThe similarity tool re-ran against the new commit. Before → after, on the three files it had flagged as newly introduced duplication:
One new entry appears in its place — PMD CPD remains at −1 clone / −28 duplicated lines vs base. Both engines pass. Why the
|
…opagation # Conflicts: # src/docs/development/upstream-map.yaml
…ew schema Master's mirror landed while this branch was open, and it changes two things under this PR's feet. Issue references must now name their repo below #1000 (#197 / confluentinc#907), enforced on added lines by .github/scripts/issue-ref-gate.js. Three of this branch's added lines were bare - two in the inflight note, one in MdcContextPropagationTest's javadoc - so the PR Checklist gate would have failed on merge. Verified by running the gate's own suspectRefs() over this PR's diff: three hits before, zero after. Also dropped "upstream confluentinc#907" for "confluentinc#907". The bare form still passes the gate, but AGENTS.md now asks new writing to name the owner rather than the role, since this fork is itself upstream to anyone who forks it. upstream-map.yaml was restructured by the mirror commit: it tracks upstream PRs only, entries moved to block style, and fork work now cross-links its mirror via fork_issue. Refitted this entry to match and renamed it issue-907-mdc-propagation -> mdc-context-propagation, since the "issue-" prefix now reads as one of the tracking-only entries that commit deleted. Safe to rename: nothing outside this branch referenced it. fork_issue: 195 is the mirror of confluentinc#907. Kept the entry rather than deleting it. AGENTS.md is in slight tension here - the manifest "tracks upstream PRs only" (this maps to an issue, not a PR), but it also says to record status: pr-open for your own fork work at each lifecycle transition. fork_issue: 195 satisfies both: the upstream issue's diagnosis and closing live on the mirror, the fork work's lifecycle lives here. upstream-map.py validate: OK, 15 entries. Refs #197, confluentinc#907. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RtNUsxokE9g2pSEjBHZqNA
… two Master's 16ac63b fixed MultiInstanceMetricsTest for awaiting a counter that LEADS the value being asserted, and generalised the rule: await the value you are about to assert, never a proxy that leads it. MdcBoundaryProbe, added earlier in this PR, had the same shape. observeCurrentThread() appended to threadsUsed and then to contextSeen, and observations() returned contextSeen. So the engine tests awaited contextSeen reaching 3 and then asserted over threadsUsed - a collection that is always one step ahead. A fourth invocation arriving concurrently could have contributed its thread name but not yet its context value, and assertObservedOnlyOn would have inspected four entries having awaited three. It could not actually lose today, because each engine test primes exactly three records, so no fourth invocation exists. That is a property of the fixture, not of the probe - exactly the "all four sites had the pattern, only one had lost the race so far" situation 16ac63b describes. Now one Observation (thread + value seen) is added atomically to one collection, so awaiting observations() bounds everything the assertions read. This is the shape MdcContextPropagationTest already used for its own @value Observation; the probe should not have diverged from it when the engine bookkeeping was extracted. Nothing loosened - the assertions are unchanged in strength and still name the threads and values seen on failure. Re-verified both Reactor and Mutiny still go RED under propagateMdc(false) after the change, with `values seen: [null, null, null]`; all 17 tests green. Refs #197. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01RtNUsxokE9g2pSEjBHZqNA
|
Claude finished @astubbs's task in 5m 49s —— View job Claude's review of #205 (re-review after commit
|
1 participant
Description
Raised as a non-blocking finding on #197 by a user in the confluentinc#907 thread (mirrored on this fork as #195): the
caller's SLF4J MDC is not propagated into the worker pool.
Confirmed on master before writing anything:
pcId(MDC_INSTANCE_ID, put inaddInstanceMDC()and inBrokerPollSystem) andoffset(MDC_WORK_CONTAINER_DESCRIPTOR).MDC.getCopyOfContextMap()andMDC.setContextMap()appear nowhere in the repository.So a caller who has established diagnostic context - a
trace_id, arequest_id, a tenant - loses all of it themoment a record crosses into the worker pool, and again crossing into vert.x, Reactor or Mutiny. The logs their
function writes cannot be correlated back to the originating request.
Where the capture and the restore live
New
MdcPropagation(coreinternal):capture()on the thread that has the context, andenter(..)returningan
AutoCloseablescope on the thread that runs the work.submitWorkToPoolInner, on the controller thread at submit timeaddVertxHooksand the web-request callback, on the worker threadFuturereact(..)'s wrapper, on the worker threadMono.fromCallableand both terminal signalsonRecord(..)'s wrapper, on the worker threadUni.deferredand both terminal signalsThe caller's own snapshot is taken in
supervisorLoop- i.e. on the thread that callspoll*(). That is the onlymoment it is reachable: none of PC's threads exist yet and the SLF4J MDC is not inheritable. The controller and
broker-poller threads then
adopt(..)it outright (they serve one PC instance for life, so there is no later taskto leak into), which is also what makes the submit-time capture pick up the caller's context.
Judgement calls
Precedence - PC's keys win.
pcIdandoffsetare applied after the caller's map at every call site, so acaller key of the same name cannot shadow them. PC's own log lines are read by those keys; a caller key shadowing
offsetwould silently corrupt PC's own diagnostics. Covered by a test that deliberately collides onpcId.On by default, with
ParallelConsumerOptions.propagateMdcas the kill switch, wired throughPCModule.mdcPropagation(). Not propagating fails silently, for everyone, and only ever gets fixed by users whoread a changelog; propagating fails visibly, as one extra key in a log line, and can be switched off. The
asymmetry decides it. Setting it
falserestores the old behaviour exactly - leak included - so it is a truerevert rather than half of one.
The one genuine risk of a default-on capture is that it is taken once, at
poll*(): a request-scoped value in theMDC at that moment gets pinned to the consumer for its whole life, which is the same "actively misleading logs"
failure this PR exists to prevent. Mitigated rather than ignored - PC logs the captured keys (never the values,
which are the user's data) at INFO on startup, so the mistake is discoverable. It is also called out on the option's
javadoc.
That INFO line is once per consumer instance, not per poll and not per record, and structurally so:
captureCallersDiagnosticContext()is called only fromsupervisorLoop, which throwsIllegalStateExceptionifpoll*()is called more than once. There is no hot path it can reach.The leak is fixed too, not just the propagation. On master, whatever the user function puts into the MDC stays
on the pooled thread for the next, unrelated, record. The restore scope fixes that as a side effect, and it is the
half of this change that matters most.
Null-safety.
getCopyOfContextMap()returnsnullfor an empty context; that is the normal path and ishandled throughout, with a test asserting no NPE and correct behaviour when the caller sets nothing at all.
Hot-path cost: measured, not assumed.
getCopyOfContextMap()returnsnullwithout allocating when thecontext is empty,
enter(null)returns a shared singleton scope, andpcIdis only ever set whensetMyId()hasbeen used. Rather than leave that as reasoning, it was measured with
ThreadMXBean.getThreadAllocatedBytesover amillion
capture()/enter()/close()cycles on logback 1.6.1:capture()on the controller threadenter()+close()on a clean worker threadSo the "nil for users who never touch the MDC" claim holds exactly - zero bytes, not merely "small". For a user who
has set MDC it is ~424 bytes per batch, against the
groupingBycollector, twoArrayLists, aPollContextInternaland aFutureTaskthatrunUserFunctionalready allocates per batch.One caveat now recorded in the inflight note: the zero is a property of logback's
MDCAdapter, which returnsnullfor an empty context. Some other SLF4J bindings return an empty map instead, which would makecapture()allocate one small map per batch. Behaviour is correct either way; only the zero-allocation figure is
binding-specific.
Modules covered, and what is not
Covered: core (worker pool), vertx (event loop), reactor (scheduler), and mutiny. Mutiny was not in
the brief, but it is a genuine third boundary of exactly the same shape as Reactor's -
runSubscriptionOn(executor)means the user's function does not run on the worker thread either - so leaving it out would have shipped a
documented inconsistency for ~5 lines of saving. It has its own test like the others.
Deliberately not covered: the operator chains of the
Publisher/Unithe user returns. We carry the contextinto the invocation of their function and into PC's terminal signal handling; following it further, onto whatever
schedulers their own chain hops to, needs Reactor's
io.micrometer:context-propagationand is the user's call, notours. Recorded in the inflight note so the next person does not treat it as an oversight.
Tests
17 tests:
MdcPropagationTest(9, the primitive itself),MdcContextPropagationTest(5, end-to-end throughthe real worker pool), and 1 each in vertx / reactor / mutiny. They assert the caller's key is visible inside the
user function, that PC's own keys survive and win a collision, that the context does not leak to a later task on the
same pooled thread, and that an empty/null context does not NPE.
The three engine tests share
MdcBoundaryProbe(core test-jar, the same routeTestConventionRulesandWireMockUtilsalready take) by composition - each already extends its own module's unit-test base, so a commonsuperclass is not available. The probe holds only what is genuinely identical: the caller key/value, the observation
bookkeeping, the MDC teardown, and the "context was visible" assertion. The per-engine assertions stay in the tests,
because they are what stops a test quietly degrading into a re-run of the core worker-pool case, and they really do
differ - Reactor asserts its scheduler thread positively (
boundedElastic), while vertx and Mutiny can onlyassert "not a
pc-thread" since their executor is caller-supplied.Every one of them is verified to be able to fail. A green test that cannot go red proves nothing, so:
expected to be empty but was: [poison_from_offset_2, poison_from_offset_3, poison_from_offset_0, poison_from_offset_1]threadsUsedhas size 1, so it cannot pass by accidentally getting a fresh thread perrecord.
maxConcurrency(1)makes the reuse deterministic rather than lucky.propagateMdc(false)(temporarily overridinginitAsyncConsumer(..)) and each failed on its own "context must be visible on the <engine>" assertion,reporting
values seen: [null, ...]. Re-verified again after the shared-probe refactor, so the extraction did notquietly defuse them. They also assert which thread observed - Reactor positively (
boundedElastic), vertx andMutiny as "not a
pc-thread" - so they cannot silently stop covering the second boundary.No existing assertion was weakened. The one failure I hit while developing (
ReactorMdcPropagationTestexpecting 4records) was my own arithmetic -
ReactorPCTestprimes a 4th record in its own@BeforeEach,ReactorUnitTestBaseprimes none - diagnosed from the surefire report (exactly 3 processed, 3 primed) rather than papered over with a
longer timeout.
Full unit suite (
bin/ci-unit-test.sh) green across all 11 modules;bin/check-copyright-headers.shclean;upstream-map.py validateOK.Refs #197, #195, confluentinc#907.
Merged up from master
Master moved a long way while this was open (the upstream-issue mirror, the repo-hygiene workflow, four SIGPIPE fixes). Merged in at
d4fed639; one conflict, inupstream-map.yaml.The conflict. #114 restructured that file: it now tracks upstream PRs only, because every upstream issue has a fork mirror, and it deleted ten tracking-only entries - one of which sat immediately beside the entry this PR adds. Took master's deletions wholesale, then refitted this PR's entry to the new shape: block style,
fork_issue: 195cross-linking the mirror of confluentinc#907, and renamedissue-907-mdc-propagation->mdc-context-propagation, since anissue-prefix now reads as one of the entries that commit removed. Safe to rename - nothing outside this branch referenced it.upstream-map.py validate: OK, 15 entries.AGENTS.md is in slight tension on whether this entry should exist at all: the manifest "tracks upstream PRs only" and this maps to an issue, but it also says to record
status: pr-openfor your own fork work at each lifecycle transition.fork_issue: 195satisfies both readings - the upstream issue's diagnosis and closing live on the mirror, this PR's lifecycle lives in the manifest. Happy to drop the entry instead if you read it the other way.The new reference gate caught three real ones. #114 also added
.github/scripts/issue-ref-gate.js, run by the PR Checklist check: below #1000 a reference must name its repo, because the fork's numbers sit entirely inside upstream's range. Ran the gate's ownsuspectRefs()over this PR's diff - three unqualified refs on added lines (two in the inflight note, one inMdcContextPropagationTest's javadoc), now zero. Also switchedupstream #907toconfluentinc#907: the bare form still passes, but AGENTS.md asks new writing to name the owner rather than the role.A latent race in this PR's own test probe, found by reading
16ac63b1. That commit fixedMultiInstanceMetricsTestfor awaiting a counter that leads the value asserted, and generalised the rule.MdcBoundaryProbehad the same shape: it appended tothreadsUsedand then tocontextSeen, and the engine tests awaitedcontextSeenbefore asserting overthreadsUsed- always one step ahead. It could not lose today, because each engine test primes exactly three records, but that is a property of the fixture rather than the probe. Now oneObservation(thread + value) is added atomically to one collection, matching whatMdcContextPropagationTestalready did. Re-verified both Reactor and Mutiny still go RED underpropagateMdc(false)afterwards.This PR pushes
AbstractParallelEoSStreamProcessor.javapast 64 KiB - 65,185 -> 67,877 bytes, i.e. 2,341 past the pipe buffer. That matters becausecheck-quarantine-owners.shused to pipe that file intogrep -qunderpipefail, which inverts its answer once the file exceeds the buffer; it had 351 bytes of headroom. Master'scb8b1182fixed it first, so nothing breaks - confirmed by running both forms against the real file: the old one reports NO-MATCH withrc=141, the new herestring reports MATCH. Worth knowing that this PR is what would have consumed that headroom, and thatbin/check-shell-sigpipe.shnow guards the class repo-wide.Checklist
MdcPropagationclass javadoc,docs/inflight/branch-mdc-context-propagation.md, and anupstream-map.yamlentry refitted to the post-mirror schema.CHANGELOG.adocdeliberately untouched, per AGENTS.md (generated at release time, not a per-PR chore)🤖 Generated with Claude Code
https://claude.ai/code/session_01RtNUsxokE9g2pSEjBHZqNA