Health-checks

[JorgenRingen]

Kafka-streams exposes a method for retrieving the current state: org.apache.kafka.streams.KafkaStreams#state.
We use this for implementing health-checks. If state is in created, rebalancing or running the application is considered healthy. Otherwise the app is considered unhealthy and the application would normally be stopped/restarted.

In PC I've implemented a health-check querying io.confluent.parallelconsumer.ParallelEoSStreamProcessor#isClosedOrFailed == false.

Would a different, more "healthcheck-like" abstraction be smart? Perhaps on the io.confluent.parallelconsumer.ParallelConsumer interface? The current method requires a cast to ParallelEoSStreamProcessor.

(not critical!)

[astubbs]

Do you have any thoughts on what metrics you might consider would indicate health? One major aspect is "progress" through the work queues. Described here: #34
Time since last progress made perhaps. Or time since last committable offset shift - per partition?

[JorgenRingen]

In kafka-streams any leaked exception from user-code or any "permanent internal errors" causes the state to change to Error -> Pending Shutdown -> Not Running. We use this for our (aggregate) application health-checks and polls KS.state. If KS is in a unhealthy state the app is typically restarted by container-platform. Sometimes a restart can make the app recover and other times there's a poison-message being consumed that causes a restart-loop and needs manual intervention.

So maybe exposing something like state on ParallelConsumer interface for application-healthchecks might be a good start and any "shutdown"-scenarios in #34 would change the state to "error".

I'm doing ParallelEosStreamProcessor#isClosedOrFailed now and it works ok, but maybe a more uniform interface wold be smart down the road.

Current spring-boot-actuator-pc-healthcheck :)

@Component
class ParallelConsumerHealthIndicator(
    parallelConsumer: ParallelStreamProcessor<String, Any>,
) : HealthIndicator {
    
    private val logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass())

    val parallelEosStreamProcessor = parallelConsumer as ParallelEoSStreamProcessor<String, Any> // notice cast

    override fun health(): Health {
        return if (!parallelEosStreamProcessor.isClosedOrFailed) {
            Health.up().build()
        } else {
            if (parallelEosStreamProcessor.failureCause != null) {
                logger.warn(
                    "Parallel-consumer is unhealthy! Failure-cause: ${parallelEosStreamProcessor.failureCause.message}",
                    parallelEosStreamProcessor.failureCause
                )
            } else {
                logger.warn("Parallel-consumer is closed or unhealthy - no failure-cause.")
            }

            Health.down().build()
        }
    }
}

[astubbs]

Draft implementation with test: #464

If you can, take a look and see if there's anything else you'd like to see. I won't get them all first time around, but that's the goal.