From a8756136ad2154f1be4d7c36005814c00278c814 Mon Sep 17 00:00:00 2001 From: mariano Date: Thu, 30 Jul 2026 23:08:46 -0500 Subject: [PATCH] =?UTF-8?q?fix:=20truthful=20health=20signal=20=E2=80=94?= =?UTF-8?q?=20HealthSnapshot=20wires=20probe,=20handle,=20and=20breaker=20?= =?UTF-8?q?together?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Handle.isHealthy() returned consumer.isRunning(), which is RUNNING || PAUSED - so a tripped circuit breaker (which pauses the consumer) read as HEALTHY during the exact incident the breaker exists for. Meanwhile ConsumerHealthController computed the real signal with zero production callers, and HttpHealthServer was an orphan no example ever started (the demo even documents HEALTH_HTTP_PORT without wiring it). - New HealthSnapshot record (running, paused, pauseSources, circuitBreaker, inFlight) with healthy() = running && breaker != OPEN - the single place the health question is answered - KPipeConsumer.health() builds it from ConsumerHealthController (first production caller for circuitBreakerState/isPaused/currentSources) - Handle.health() added; DefaultHandle.isHealthy() delegates to snapshot.healthy() - Demo app now starts HttpHealthServer from env, suppliers bound to handle.health() - the documented /health endpoint exists for real, and reports breaker state - getPartitionState labeled as a test observation point (audit finding 16) - KPipeCircuitBreakerIntegrationTest extended: during OPEN, snapshot must report running=true, paused=true, source=CIRCUIT_BREAKER, healthy()=false --- .../github/eschizoid/kpipe/demo/DemoApp.java | 16 ++++++++- .../github/eschizoid/kpipe/DefaultHandle.java | 11 +++++- .../io/github/eschizoid/kpipe/Handle.java | 8 +++++ .../eschizoid/kpipe/HandleDefaultsTest.java | 8 +++++ .../kpipe/consumer/HealthSnapshot.java | 35 +++++++++++++++++++ .../kpipe/consumer/KPipeConsumer.java | 27 ++++++++------ .../kpipe/consumer/KafkaOffsetManager.java | 3 ++ .../KPipeCircuitBreakerIntegrationTest.java | 10 ++++++ 8 files changed, 106 insertions(+), 12 deletions(-) create mode 100644 lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/HealthSnapshot.java diff --git a/examples/demo/src/main/java/io/github/eschizoid/kpipe/demo/DemoApp.java b/examples/demo/src/main/java/io/github/eschizoid/kpipe/demo/DemoApp.java index 087b9be7..12937c01 100644 --- a/examples/demo/src/main/java/io/github/eschizoid/kpipe/demo/DemoApp.java +++ b/examples/demo/src/main/java/io/github/eschizoid/kpipe/demo/DemoApp.java @@ -8,6 +8,7 @@ import io.github.eschizoid.kpipe.format.avro.AvroFormat; import io.github.eschizoid.kpipe.format.json.JsonConsoleSink; import io.github.eschizoid.kpipe.format.protobuf.ProtobufFormat; +import io.github.eschizoid.kpipe.health.HttpHealthServer; import io.github.eschizoid.kpipe.metrics.otel.OtelConsumerMetrics; import io.github.eschizoid.kpipe.registry.Operators; import io.github.eschizoid.kpipe.schemaregistry.confluent.ConfluentSchemaResolver; @@ -50,7 +51,20 @@ static void main() { try (final var app = new DemoApp(config, avroFormat, protoFormat)) { LOGGER.log(Level.INFO, "Demo application started — JSON/Avro/Protobuf routes via KPipe.multi"); - app.handle.awaitShutdown(); + // Liveness probe wired to the handle's health snapshot: reports unhealthy when the + // circuit breaker is OPEN, not merely when the process has exited. Enabled/configured via + // HEALTH_* environment variables (see HealthConfig); disabled = empty Optional, no server. + final var healthServer = HttpHealthServer.fromEnv( + () -> app.handle.health().healthy(), + () -> app.handle.health().inFlight(), + () -> app.handle.health().paused(), + "demo-app" + ); + try { + app.handle.awaitShutdown(); + } finally { + healthServer.ifPresent(HttpHealthServer::close); + } } catch (final Exception e) { LOGGER.log(Level.ERROR, "Fatal error in demo application", e); System.exit(1); diff --git a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultHandle.java b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultHandle.java index bb871e87..8294df9e 100644 --- a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultHandle.java +++ b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/DefaultHandle.java @@ -1,5 +1,6 @@ package io.github.eschizoid.kpipe; +import io.github.eschizoid.kpipe.consumer.HealthSnapshot; import io.github.eschizoid.kpipe.consumer.KPipeConsumer; import java.time.Duration; import java.util.List; @@ -16,7 +17,15 @@ record DefaultHandle(KPipeConsumer consumer) implements Handle { @Override public boolean isHealthy() { - return consumer.isRunning(); + // Delegates to the snapshot's definition: running AND breaker not OPEN. `isRunning()` alone + // reads true while a tripped circuit breaker has the consumer PAUSED — the one incident a + // liveness probe must see. + return consumer.health().healthy(); + } + + @Override + public HealthSnapshot health() { + return consumer.health(); } @Override diff --git a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/Handle.java b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/Handle.java index ab5ce6b1..4fdc0fa3 100644 --- a/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/Handle.java +++ b/lib/kpipe-api/src/main/java/io/github/eschizoid/kpipe/Handle.java @@ -1,5 +1,6 @@ package io.github.eschizoid.kpipe; +import io.github.eschizoid.kpipe.consumer.HealthSnapshot; import java.time.Duration; import java.util.List; import java.util.Map; @@ -15,6 +16,13 @@ public interface Handle extends AutoCloseable { /// @return true when the consumer is running and the configured health check passes boolean isHealthy(); + /// Full health snapshot: running/paused state, active pause sources, circuit-breaker state, + /// and in-flight count. Wire liveness probes (e.g. `HttpHealthServer`) to this rather than + /// polling `metrics()`. + /// + /// @return an immutable point-in-time health view + HealthSnapshot health(); + /// Returns an unmodifiable snapshot of the consumer's metrics. All values are counters or /// gauges represented as `Long`. Returns an empty map when metrics are disabled on the /// underlying consumer. diff --git a/lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/HandleDefaultsTest.java b/lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/HandleDefaultsTest.java index 054b99b3..a6d5878b 100644 --- a/lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/HandleDefaultsTest.java +++ b/lib/kpipe-api/src/test/java/io/github/eschizoid/kpipe/HandleDefaultsTest.java @@ -3,9 +3,12 @@ import static java.nio.charset.StandardCharsets.UTF_8; import static org.junit.jupiter.api.Assertions.*; +import io.github.eschizoid.kpipe.consumer.CircuitBreakerState; +import io.github.eschizoid.kpipe.consumer.HealthSnapshot; import java.time.Duration; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.atomic.AtomicReference; import org.junit.jupiter.api.Test; @@ -26,6 +29,11 @@ public boolean isHealthy() { return true; } + @Override + public HealthSnapshot health() { + return new HealthSnapshot(true, false, Set.of(), CircuitBreakerState.CLOSED, 0L); + } + @Override public Map metrics() { return Map.of(); diff --git a/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/HealthSnapshot.java b/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/HealthSnapshot.java new file mode 100644 index 00000000..ddfd9014 --- /dev/null +++ b/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/HealthSnapshot.java @@ -0,0 +1,35 @@ +package io.github.eschizoid.kpipe.consumer; + +import java.util.Set; + +/// Point-in-time health view of a running consumer, taken via [KPipeConsumer#health()]. +/// +/// This is the one place the "is this consumer healthy" question is answered. A consumer that +/// paused itself because the circuit breaker tripped is alive but NOT healthy — the breaker +/// pauses consumption precisely because downstream is failing — so [#healthy()] is the value a +/// liveness/readiness probe should serve, not `running` alone. +/// +/// @param running the consumer thread is alive (RUNNING or PAUSED state) +/// @param paused consumption is currently paused (any source) +/// @param pauseSources names of the active pause sources (`MANUAL`, `BACKPRESSURE`, +/// `CIRCUIT_BREAKER`); empty when not paused +/// @param circuitBreaker current circuit-breaker state; `CLOSED` when no breaker is configured +/// @param inFlight records currently in flight (dispatched + buffered batch records) +public record HealthSnapshot( + boolean running, + boolean paused, + Set pauseSources, + CircuitBreakerState circuitBreaker, + long inFlight +) { + public HealthSnapshot { + pauseSources = Set.copyOf(pauseSources); + } + + /// `true` when the consumer is running and the circuit breaker is not OPEN. Backpressure or + /// manual pauses do not make a consumer unhealthy — they are normal flow control — but an + /// OPEN breaker means downstream is failing and consumption is deliberately halted. + public boolean healthy() { + return running && circuitBreaker != CircuitBreakerState.OPEN; + } +} diff --git a/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KPipeConsumer.java b/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KPipeConsumer.java index 349bf575..3f7e65ff 100644 --- a/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KPipeConsumer.java +++ b/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KPipeConsumer.java @@ -1,6 +1,5 @@ package io.github.eschizoid.kpipe.consumer; -import io.github.eschizoid.kpipe.consumer.config.AppConfig; import io.github.eschizoid.kpipe.metrics.ConsumerMetricKeys; import io.github.eschizoid.kpipe.metrics.ConsumerMetrics; import io.github.eschizoid.kpipe.metrics.KPipeMetricsReporter; @@ -8,8 +7,6 @@ import io.github.eschizoid.kpipe.producer.tracing.Tracer; import io.github.eschizoid.kpipe.registry.MessagePipeline; import io.github.eschizoid.kpipe.registry.Result; -import io.github.eschizoid.kpipe.sink.BatchPolicy; -import io.github.eschizoid.kpipe.sink.BatchSink; import java.lang.System.Logger; import java.lang.System.Logger.Level; import java.nio.channels.ClosedByInterruptException; @@ -20,14 +17,11 @@ import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.LockSupport; -import java.util.function.Function; -import java.util.function.Supplier; import java.util.stream.Collectors; import org.apache.kafka.clients.consumer.*; import org.apache.kafka.clients.producer.*; import org.apache.kafka.common.errors.InterruptException; import org.apache.kafka.common.errors.WakeupException; -import org.apache.kafka.common.serialization.ByteArrayDeserializer; /// A functional-style Kafka consumer that processes records using a provided function. /// @@ -128,7 +122,8 @@ public class KPipeConsumer implements AutoCloseable { /// Composes pause arbitration + backpressure decision + circuit-breaker state machine. The /// underlying decision modules ([BackpressureController], [CircuitBreakerController]) remain /// public + testable on their own; this controller owns the side-effect choreography and - /// dispatches pause transitions through a PauseLifecycleHook that points back at `internalPause` / + /// dispatches pause transitions through a PauseLifecycleHook that points back at `internalPause` + // / /// `internalResume`, and metric events through a HealthMetricsObserver bound to the counters. private final ConsumerHealthController health; @@ -178,7 +173,6 @@ public static KPipeConsumerBuilder builder() { return new KPipeConsumerBuilder(); } - /// Creates a new KPipeConsumer using the provided builder. /// /// @param builder the builder containing the consumer configuration @@ -454,9 +448,11 @@ public void onBatchFailure(final ConsumerRecord record, final Ex otelMetrics.recordProcessingError(record.topic()); health.recordOutcome(false); LOGGER.log(Level.WARNING, () -> "Batch failure for record at offset " + record.offset(), cause); - // LOCKSTEP: mirror of the per-record path's DLQ-or-mark block in handleProcessingError — + // LOCKSTEP: mirror of the per-record path's DLQ-or-mark block in handleProcessingError + // — // mark the offset only after a successful DLQ send; a failed send leaves it pending so - // the record is reprocessed, never dropped. Deliberately duplicated (the paths differ on + // the record is reprocessed, never dropped. Deliberately duplicated (the paths differ + // on // span handling, retry counts, and circuit-breaker ordering — recordOutcome runs BEFORE // this block here, AFTER it on the per-record path). Any change here must be mirrored // there; DlqTerminalContractTest asserts both paths cell-for-cell and fails on drift. @@ -876,6 +872,17 @@ public boolean isRunning() { return s == ConsumerState.RUNNING || s == ConsumerState.PAUSED; } + /// Point-in-time health snapshot: running/paused state, active pause sources, circuit-breaker + /// state, and in-flight count. This is the source of truth for liveness probes — see + /// [HealthSnapshot#healthy()] for why `isRunning()` alone is the wrong probe signal (a + /// breaker-tripped consumer is alive but deliberately not consuming). + /// + /// @return an immutable snapshot of this consumer's health + public HealthSnapshot health() { + final var sources = health.currentSources().stream().map(Enum::name).collect(Collectors.toUnmodifiableSet()); + return new HealthSnapshot(isRunning(), health.isPaused(), sources, health.circuitBreakerState(), totalInFlight()); + } + /// Atomically transitions from any active state (RUNNING or PAUSED) to CLOSING. /// Uses a single-read CAS to avoid the double-CAS window. /// diff --git a/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KafkaOffsetManager.java b/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KafkaOffsetManager.java index 2d98f96f..275c8050 100644 --- a/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KafkaOffsetManager.java +++ b/lib/kpipe-consumer/src/main/java/io/github/eschizoid/kpipe/consumer/KafkaOffsetManager.java @@ -296,6 +296,9 @@ private boolean performCommit(final Map offse /// /// @param partition The partition to get state for /// @return the partition's offset-tracking state + /// **Test observation point.** No production caller — jcstress and property suites assert on + /// the commit frontier through this window. Not intended for operational dashboards; wire + /// [KPipeConsumer#health()] or `getStatistics()` for those. public PartitionState getPartitionState(final TopicPartition partition) { return ledger.partitionState(partition, state.get()); } diff --git a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KPipeCircuitBreakerIntegrationTest.java b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KPipeCircuitBreakerIntegrationTest.java index 5d26916b..802299c6 100644 --- a/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KPipeCircuitBreakerIntegrationTest.java +++ b/lib/kpipe-consumer/src/test/java/io/github/eschizoid/kpipe/consumer/KPipeCircuitBreakerIntegrationTest.java @@ -79,6 +79,16 @@ void sustainedFailuresTripBreakerAndPauseConsumer() throws InterruptedException "exactly one trip should have been recorded" ); + // The health snapshot must tell the truth during the incident: the consumer thread is + // alive (running) but the breaker is OPEN, so healthy() is false. isRunning() alone reads + // true here - the exact false-positive a liveness probe must not serve. + final var snapshot = consumer.health(); + assertTrue(snapshot.running(), "consumer thread is alive while the breaker is open"); + assertTrue(snapshot.paused(), "breaker pause must be visible in the snapshot"); + assertEquals(CircuitBreakerState.OPEN, snapshot.circuitBreaker(), "snapshot must expose the OPEN breaker"); + assertTrue(snapshot.pauseSources().contains("CIRCUIT_BREAKER"), "pause source must name the breaker"); + assertTrue(!snapshot.healthy(), "an OPEN breaker must make the consumer unhealthy"); + consumer.close(); }