fix(consumer): close resources on constructor-failure paths

[eschizoid]

What

Fixes three resource-leak risks surfaced by the resource-leak audit, all on construction paths where a partial init would otherwise strand already-opened resources for the JVM lifetime.

1. (MEDIUM) KPipeConsumer constructor leaks resources if a later field init throws

The constructor opens, in order: the Kafka consumer, the dispatcher (PARALLEL owns a virtual-thread executor), an optional built-in DLQ KPipeProducer, an optional scheduler, the batch wrappers, and a JVM shutdown hook. If anything after the first open threw — e.g. addShutdownHook while the JVM is already shutting down (IllegalStateException), or a user-supplied ConsumerMetrics/Tracer/CircuitBreakerController whose construction work throws — every already-opened resource leaked.

The constructor body is now wrapped so any throwable triggers closePartiallyConstructed(t), which closes whatever was opened in reverse of the open order (shutdown hook → scheduler → producer → batch wrappers → dispatcher → Kafka consumer), suppressing any cleanup failure onto the original throwable, then rethrows.

Ownership mirrors the existing teardown (releaseConstructedResources / the consumer-thread finally): the Kafka consumer and the DLQ producer are closed regardless of whether they were supplied via withConsumer(...) / withKafkaProducer(...) or built internally, because once handed to the consumer they are consumer-owned. The helper reads fields off this (a partially-constructed instance leaves not-yet-assigned fields at their default null), so every step is null-guarded. The whitespace-ignoring diff for this file is +71 lines (the try, the catch, and the helper); the rest of the large diff is pure re-indentation of the wrapped body.

This is the load-bearing fix.

2. (LOW) HttpHealthServer constructor leaks bound socket if createContext throws

HttpServer.create(addr, 0) binds the socket; if the subsequent createContext(path, ...) throws, the socket leaked. Now wrapped: try { createContext(...) } catch (RuntimeException e) { server.stop(0); throw e; }.

3. (LOW) HttpHealthServer.fromEnv Javadoc said "started" but never called start()

The Javadoc promised "a started HttpHealthServer instance", but fromEnv never started the server. The only caller is a test exercising the disabled path, so nothing started it on the caller side — the doc claim was a real bug. fromEnv now calls server.start() before returning, matching the documented contract (no double-start: start() is idempotent via a CAS guard).

Deferred

Scheduler awaitTermination finding (#4 in the audit) was considered and deliberately deferred. It is purely defensive — all scheduled tasks are cancelled before shutdownNow() today, so there is nothing to await — and adding an awaitTermination would only add shutdown latency for no behavioral gain.

Tests

• KPipeConsumerMockingTest.constructorClosesKafkaConsumerWhenLaterInitStepThrows — supplies a real MockConsumer via withConsumer(...) plus a withOffsetManagerProvider(...) that throws (the provider runs after both the Kafka consumer and the dispatcher are open), asserts the thrown exception propagates unchanged, and asserts mockConsumer.closed() — proving the load-bearing Kafka-consumer close.
• HttpHealthServerTest.shouldReturnStartedServerFromEnv — mocks the HealthConfig env reads to an ephemeral port and hits the endpoint over HTTP without calling start(), proving fromEnv started it.

No test was added for #2 (createContext failure): triggering it cleanly requires spying/subclassing HttpServer since createContext only throws on a duplicate/invalid path that normalizePath already prevents — a brittle test for a small, reviewable fix. Relying on review there.

Verification

./gradlew :lib:kpipe-consumer:test — 265 passing; the only 2 failures are the known DockerClientProviderStrategy Testcontainers initialization errors (environmental, no Docker in this sandbox).

[eschizoid]

@copilot please review

[codecov]

Codecov Report

❌ Patch coverage is 74.63768% with 35 lines in your changes missing coverage. Please review.
✅ Project coverage is 78.36%. Comparing base (4ccac98) to head (099b25e).
⚠️ Report is 16 commits behind head on main.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
...github/eschizoid/kpipe/consumer/KPipeConsumer.java	75.38%	21 Missing and 11 partials ⚠️
...ithub/eschizoid/kpipe/health/HttpHealthServer.java	62.50%	3 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main     #209      +/-   ##
============================================
+ Coverage     77.82%   78.36%   +0.54%     
- Complexity      732      755      +23     
============================================
  Files            68       66       -2     
  Lines          2800     2866      +66     
  Branches        346      366      +20     
============================================
+ Hits           2179     2246      +67     
+ Misses          462      454       -8     
- Partials        159      166       +7     

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