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(); }