refactor(consumer): fold KPipeRunner + MessageTracker into KPipeConsumer by eschizoid · Pull Request #107 · eschizoid/kpipe

eschizoid · 2026-05-15T22:50:09Z

Summary

The 1.12 lifecycle path was Handle → KPipeRunner → KPipeConsumer → MessageTracker — three lifecycle layers plus an in-flight-counting helper, each exposing close / isHealthy / awaitShutdown / inflight concerns. This PR collapses everything onto KPipeConsumer directly.

Reopens a previously-rejected PLAN.md decision — the rejection ("the split has architectural meaning") didn't survive the docs sweep, which exposed how much teaching tax the three-layer story added. PLAN.md will be updated in the final architecture-improvements branch landing.

New on KPipeConsumer

waitForInFlightDrain(Duration) — replaces MessageTracker.waitForCompletion(), uses live totalInFlight() (includes buffered batch records) instead of re-deriving from metrics
awaitShutdown() / awaitShutdown(Duration) — backed by CountDownLatch, fires on both close() and the consumer thread's terminal finally
shutdownGracefully(Duration) — pause → bounded drain → close, returns whether drain finished
Builder additions: withMetricsReporters(...), withMetricsInterval(Duration), withShutdownHook(boolean) — daemon thread + JVM hook live on the consumer now

Deleted

KPipeRunner.java + KPipeRunnerTest.java
MessageTracker.java + MessageTrackerTest.java
KPipeConsumer.createMessageTracker() (was used internally only)

Migrated

DefaultHandle now records KPipeConsumer directly. DefaultSink / DefaultBatchSink / MultiBuilder call consumer.start(). README §"KPipeRunner" rewritten as §"Consumer lifecycle"; full-app example migrated. CLAUDE.md §13 + §17 capability table updated.

Stack

Stacks on top of #106. Land #105 → #106 → this one.

Test plan

lib/kpipe-consumer, lib/kpipe-api, lib/kpipe-core unit tests green (BUILD SUCCESSFUL in 5m 35s on commit cdfe1ff)
Verify a long-running consumer with .withMetricsReporters(...) emits a periodic log line at the configured interval
Verify a JVM SIGTERM with .withShutdownHook(true) drains in-flight records before exit

codecov · 2026-05-15T23:00:42Z

Codecov Report

❌ Patch coverage is 32.63158% with 64 lines in your changes missing coverage. Please review.
✅ Project coverage is 70.34%. Comparing base (d45e4cf) to head (c94b696).

Files with missing lines	Patch %	Lines
...rc/main/java/org/kpipe/consumer/KPipeConsumer.java	26.19%	56 Missing and 6 partials ⚠️
...ipe-api/src/main/java/org/kpipe/DefaultHandle.java	60.00%	2 Missing ⚠️

Additional details and impacted files

@@                     Coverage Diff                     @@
##             arch/metrics-reporter     #107      +/-   ##
===========================================================
- Coverage                    73.33%   70.34%   -3.00%     
+ Complexity                     630      589      -41     
===========================================================
  Files                           65       63       -2     
  Lines                         2453     2347     -106     
  Branches                       299      300       +1     
===========================================================
- Hits                          1799     1651     -148     
- Misses                         509      553      +44     
+ Partials                       145      143       -2

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

The 1.12 lifecycle path was Handle → KPipeRunner → KPipeConsumer → MessageTracker — three lifecycle layers + one in-flight-counting helper, each exposing close / isHealthy / awaitShutdown / inflight concerns. 1.13.0 collapses everything onto KPipeConsumer directly. New on KPipeConsumer: - waitForInFlightDrain(Duration) — replaces MessageTracker.waitForCompletion(), uses the live totalInFlight() counter (which includes buffered batch records) rather than re-deriving received - processed - errors from metrics - awaitShutdown() / awaitShutdown(Duration) — backed by a CountDownLatch counted down at the end of close() and in the consumer thread's terminal finally (so an uncaught exception also unblocks waiters) - shutdownGracefully(Duration) — pause → drain (bounded) → close, returns whether the drain finished cleanly - withMetricsReporters(Collection<KPipeMetricsReporter>) / withMetricsInterval(Duration) on the builder — adds a daemon thread that fires each reporter at the configured cadence while the consumer is alive - withShutdownHook(boolean) — installs Runtime.getRuntime().addShutdownHook(close) Deleted: - KPipeRunner.java + KPipeRunnerTest.java - MessageTracker.java + MessageTrackerTest.java - createMessageTracker() from KPipeConsumer (was used internally only) Migrated: - DefaultHandle is now a record over KPipeConsumer alone - DefaultSink / DefaultBatchSink / MultiBuilder call consumer.start() directly - Handle / Sink javadoc, KPipeConsumer class-level docs - README §"KPipeRunner" rewritten as §"Consumer lifecycle"; full-app example migrated; module catalog rows updated - CLAUDE.md §13 + §17 capability table (Lifecycle handle, periodic metrics reporting, in-flight drain rows) Per §16 — no deprecation shim. All callers migrated in this commit.

… + awaitShutdown rethrows InterruptedException Review follow-ups from PR #107: 1. close() on a CREATED-state consumer (built but never started) used to exit via the transitionToClosing() early-return without ever calling shutdownLatch.countDown(). Any awaitShutdown() caller blocked forever. Now close() handles CREATED → CLOSED directly and releases the latch. (HIGH #1 from the code review.) 2. KPipeConsumer.awaitShutdown() previously caught InterruptedException internally and returned false / void. That contradicted the Handle.awaitShutdown() interface declaration of `throws InterruptedException`, silently dropping the exception that the contract told callers to handle. Now the no-arg awaitShutdown actually throws (after restoring the interrupt flag); the bounded awaitShutdown(Duration) keeps its boolean return for the interruption case. (HIGH #2 from the code review.)

…lose() docs Review follow-ups from PR #107 (MEDIUM): 1. JVM shutdown hook was added in the constructor but never removed. For a process that builds and closes many consumers, hooks accumulated in the shutdown-hook set for the JVM's lifetime — each holding a strong reference to the consumer. close() now removes its hook via Runtime.removeShutdownHook, ignoring the IllegalStateException that fires when removing during shutdown (the hook is currently running on the very path that called us). 2. shutdownGracefully(Duration) used a plain state.get() == CLOSED read and fell through to pause()/close() on every non-CLOSED state. Under race with another thread already mid-shutdown (state = CLOSING), it would call pause on a closing consumer and return a misleading 'drained = true'. Now it snapshots the state once and treats CLOSING as 'await the other shutdown, report its drain status' rather than starting a new shutdown sequence. 3. close()'s Javadoc still listed 'Creating a message tracker' as step 2 — a dead reference from before the KPipeRunner+MessageTracker fold. Replaced the step list to reflect what close() actually does today (CREATED→CLOSED fast-path, stop metrics thread, pause + drain, etc.).

…-close Review follow-up from PR #107 (MEDIUM): DefaultSink, DefaultBatchSink, and MultiBuilder each implemented the same "call consumer.start(); on RuntimeException close + addSuppressed + rethrow; return new DefaultHandle(consumer)" pattern. Promoted that to a static `DefaultHandle.startAndWrap(KPipeConsumer<byte[]>)` so the three sites collapse to one-liners. The error-handling logic is now in one place — fix once, fixed everywhere.

…/MEDIUM/LOW review fixes (#113) * refactor(consumer): collapse Processor+Sink reporters into EntryMetricsReporter ProcessorMetricsReporter and SinkMetricsReporter were structurally identical records — same fields, same iterate-and-format loop, only the log-message prefix differed ("Processor" vs "Sink"). 1.13.0 collapses them into one parametric EntryMetricsReporter that carries the prefix explicitly and exposes namespace-specific factory methods: EntryMetricsReporter.forProcessors(registry) EntryMetricsReporter.forProcessors(registry, keys) EntryMetricsReporter.forSinks(registry) EntryMetricsReporter.forSinks(registry, keys) ConsumerMetricsReporter is left alone — it computes an aggregate snapshot (uptime, per-second rates, backpressure totals), not a per-entry loop, so the shapes don't unify. Tests: ProcessorMetricsReporterTest + SinkMetricsReporterTest collapsed into EntryMetricsReporterTest with named test groups for each namespace plus the shared empty/exception/fluent behavior. CLAUDE.md §9 and §10 updated; README example bumped. Per §16 no-deprecation — callers migrated in this commit. * refactor(consumer): fold KPipeRunner + MessageTracker into KPipeConsumer The 1.12 lifecycle path was Handle → KPipeRunner → KPipeConsumer → MessageTracker — three lifecycle layers + one in-flight-counting helper, each exposing close / isHealthy / awaitShutdown / inflight concerns. 1.13.0 collapses everything onto KPipeConsumer directly. New on KPipeConsumer: - waitForInFlightDrain(Duration) — replaces MessageTracker.waitForCompletion(), uses the live totalInFlight() counter (which includes buffered batch records) rather than re-deriving received - processed - errors from metrics - awaitShutdown() / awaitShutdown(Duration) — backed by a CountDownLatch counted down at the end of close() and in the consumer thread's terminal finally (so an uncaught exception also unblocks waiters) - shutdownGracefully(Duration) — pause → drain (bounded) → close, returns whether the drain finished cleanly - withMetricsReporters(Collection<KPipeMetricsReporter>) / withMetricsInterval(Duration) on the builder — adds a daemon thread that fires each reporter at the configured cadence while the consumer is alive - withShutdownHook(boolean) — installs Runtime.getRuntime().addShutdownHook(close) Deleted: - KPipeRunner.java + KPipeRunnerTest.java - MessageTracker.java + MessageTrackerTest.java - createMessageTracker() from KPipeConsumer (was used internally only) Migrated: - DefaultHandle is now a record over KPipeConsumer alone - DefaultSink / DefaultBatchSink / MultiBuilder call consumer.start() directly - Handle / Sink javadoc, KPipeConsumer class-level docs - README §"KPipeRunner" rewritten as §"Consumer lifecycle"; full-app example migrated; module catalog rows updated - CLAUDE.md §13 + §17 capability table (Lifecycle handle, periodic metrics reporting, in-flight drain rows) Per §16 — no deprecation shim. All callers migrated in this commit. * fix(consumer): close() releases shutdownLatch for unstarted consumers + awaitShutdown rethrows InterruptedException Review follow-ups from PR #107: 1. close() on a CREATED-state consumer (built but never started) used to exit via the transitionToClosing() early-return without ever calling shutdownLatch.countDown(). Any awaitShutdown() caller blocked forever. Now close() handles CREATED → CLOSED directly and releases the latch. (HIGH #1 from the code review.) 2. KPipeConsumer.awaitShutdown() previously caught InterruptedException internally and returned false / void. That contradicted the Handle.awaitShutdown() interface declaration of `throws InterruptedException`, silently dropping the exception that the contract told callers to handle. Now the no-arg awaitShutdown actually throws (after restoring the interrupt flag); the bounded awaitShutdown(Duration) keeps its boolean return for the interruption case. (HIGH #2 from the code review.) * fix(consumer): shutdown-hook leak + shutdownGracefully race + stale close() docs Review follow-ups from PR #107 (MEDIUM): 1. JVM shutdown hook was added in the constructor but never removed. For a process that builds and closes many consumers, hooks accumulated in the shutdown-hook set for the JVM's lifetime — each holding a strong reference to the consumer. close() now removes its hook via Runtime.removeShutdownHook, ignoring the IllegalStateException that fires when removing during shutdown (the hook is currently running on the very path that called us). 2. shutdownGracefully(Duration) used a plain state.get() == CLOSED read and fell through to pause()/close() on every non-CLOSED state. Under race with another thread already mid-shutdown (state = CLOSING), it would call pause on a closing consumer and return a misleading 'drained = true'. Now it snapshots the state once and treats CLOSING as 'await the other shutdown, report its drain status' rather than starting a new shutdown sequence. 3. close()'s Javadoc still listed 'Creating a message tracker' as step 2 — a dead reference from before the KPipeRunner+MessageTracker fold. Replaced the step list to reflect what close() actually does today (CREATED→CLOSED fast-path, stop metrics thread, pause + drain, etc.). * refactor(metrics): drop null-default trick + add LOGGING constant (#108) Review follow-ups from PR #106 (MEDIUM): 1. The canonical constructor silently replaced a null `reporter` with `this::logMetrics`, breaking record transparency — `new EntryMetricsReporter(..., null).reporter()` returned a non-null method ref, not null. Now the constructor rejects null; callers asking for log-based output use the new public `LOGGING` constant (a `Consumer<String>` that logs each line at INFO). 2. The four static factories (`forProcessors`, `forProcessors(_, keys)`, `forSinks`, `forSinks(_, keys)`) now wire `LOGGING` explicitly instead of relying on the null trick — same call sites, but the record's transparency contract is honored. Tests: `forProcessors_defaultsToLoggerWhenConsumerIsNull` renamed and sharpened to assert the wiring uses the public `LOGGING` constant. Added `canonicalConstructorRejectsNullReporter` for the new invariant. * refactor(api): extract DefaultHandle.startAndWrap to dedupe start-and-close Review follow-up from PR #107 (MEDIUM): DefaultSink, DefaultBatchSink, and MultiBuilder each implemented the same "call consumer.start(); on RuntimeException close + addSuppressed + rethrow; return new DefaultHandle(consumer)" pattern. Promoted that to a static `DefaultHandle.startAndWrap(KPipeConsumer<byte[]>)` so the three sites collapse to one-liners. The error-handling logic is now in one place — fix once, fixed everywhere. * refactor: extract backpressure-default constants + clarify dual-CLOSED path (#110) Review follow-ups (LOW): 1. `KPipeConsumer.Builder.withBackpressure()` (no-arg) and the constructor's no-controller fallback each hard-coded `(10_000, 7_000)`. Two sites, same numbers, drift risk. Moved to two public constants on `BackpressureController` — `DEFAULT_HIGH_WATERMARK` and `DEFAULT_LOW_WATERMARK` — so the default lives in one place. 2. The consumer thread's terminal `finally` and `close()` both run `state.set(CLOSED) + shutdownLatch.countDown()`. Both paths are idempotent and the countDown-past-zero is a no-op, but it wasn't obvious without context why the duplication is safe and necessary. Added a brief inline comment explaining that the thread-finally path catches the uncaught-exception case where close() is never called. Skipped the `registerEnum` → `registerOperatorEnum` rename: §7 of CLAUDE.md explicitly documents the asymmetry (operators unsuffixed = primary surface, sinks suffixed = secondary), so renaming would break that intentional convention. * docs(plan): add Result<T> sealed type as P5 2.0 candidate Track the §12 type-level enforcement idea — make filter vs fail distinct types rather than enforcing 'null = filter, throw = fail' through discipline. Sealed Result<T> { Passed | Filtered | Failed } upgrades the convention to a compile-time guarantee. Deferred until the 2.0 window because (a) it's a hard API break on MessagePipeline return type, (b) the per-record allocation cost needs JMH evidence, and (c) §12 has been holding fine under current discipline.

eschizoid force-pushed the arch/metrics-reporter branch from 95df7b7 to 95a5e85 Compare May 15, 2026 23:07

eschizoid force-pushed the arch/runner-tracker-fold branch from cdfe1ff to 74dec25 Compare May 15, 2026 23:07

eschizoid added 2 commits May 15, 2026 18:19

eschizoid force-pushed the arch/metrics-reporter branch from 95a5e85 to d45e4cf Compare May 15, 2026 23:19

eschizoid force-pushed the arch/runner-tracker-fold branch from 74dec25 to c94b696 Compare May 15, 2026 23:19

eschizoid merged commit a27fabd into arch/metrics-reporter May 15, 2026
3 of 5 checks passed

eschizoid mentioned this pull request May 15, 2026

recovery: land #106 + #107 + #108 + #109 + #110 + #112 content + HIGH/MEDIUM/LOW review fixes #113

Merged

2 tasks

eschizoid deleted the arch/runner-tracker-fold branch May 16, 2026 12:46

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

refactor(consumer): fold KPipeRunner + MessageTracker into KPipeConsumer#107

refactor(consumer): fold KPipeRunner + MessageTracker into KPipeConsumer#107
eschizoid merged 2 commits into
arch/metrics-reporterfrom
arch/runner-tracker-fold

eschizoid commented May 15, 2026 •

edited

Loading

Uh oh!

codecov Bot commented May 15, 2026 •

edited

Loading

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

Conversation

eschizoid commented May 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

New on KPipeConsumer

Deleted

Migrated

Stack

Test plan

Uh oh!

codecov Bot commented May 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

1 participant

eschizoid commented May 15, 2026 •

edited

Loading

codecov Bot commented May 15, 2026 •

edited

Loading