refactor(api): extract DefaultHandle.startAndWrap to dedupe start-and-close#112
Closed
eschizoid wants to merge 2 commits into
Closed
refactor(api): extract DefaultHandle.startAndWrap to dedupe start-and-close#112eschizoid wants to merge 2 commits into
eschizoid wants to merge 2 commits into
Conversation
This was referenced May 15, 2026
…-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.
…D 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.
eschizoid
force-pushed
the
arch/entry-reporter-cleanup
branch
from
5b4a5a4 to
86899b6
Compare
eschizoid
force-pushed
the
arch/facade-start-helper
branch
from
06961bd to
2462b72
Compare
Owner
Author
|
Superseded by recovery PR — see #109 comment. |
Merged
2 tasks
eschizoid
added a commit
that referenced
this pull request
May 16, 2026
…/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.
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
MEDIUM-priority fix from the stack review:
DefaultSink,DefaultBatchSink, andMultiBuildereach implemented the same "callconsumer.start(); onRuntimeExceptionclose +addSuppressed+ rethrow; returnnew DefaultHandle(consumer)" pattern. Three identical 12-line blocks. Promoted to a staticDefaultHandle.startAndWrap(KPipeConsumer<byte[]>)so the three sites collapse to one-liners and the error-handling logic lives in one place.Stack
Stacks on top of #108 (entry-reporter-cleanup).
Test plan
lib/kpipe-apicompile + unit tests green locally