fixes #71: Feature: Health check and metrics

[astubbs]

fixes:

• Health-checks #71

related:

• Micrometer metrics #27

Checklist

• Documentation (if applicable)
• Changelog

blocked by:

• fix: Improvements to encoding ranges (int vs long) #439

[JorgenRingen]

Nice 👍 Metrics will definitively be useful. Didn't quite get how a health-check should be implemented? parallelConsumer.calculateMetrics().getPollerMetrics().getState()?

Or would it be possible to expose io.confluent.parallelconsumer.internal.AbstractParallelEoSStreamProcessor#state public?

Just a question regarding the class-hierarchy:
The way we've implemented it now is to instantiate through ParallelStreamProcessor.createEosStreamProcessor which returns ParallelStreamProcessor. This requires a somewhat unpleasant cast to be able to retrieve the metrics, failure-cause and perhaps state:

ParallelStreamProcessor<String, String> parallelStreamProcessor = ParallelStreamProcessor.createEosStreamProcessor(null);

// needs to cast to get metrics and failure-cause
ParallelEoSStreamProcessor<String, String> parallelEoSStreamProcessor = (ParallelEoSStreamProcessor<String, String>) parallelStreamProcessor;
PCMetrics metrics = parallelEoSStreamProcessor.calculateMetrics();
Exception failureCause = parallelEoSStreamProcessor.getFailureCause();

// whould be nice to retrieve state directly
// State state = parallelEoSStreamProcessor.getState()

Perhaps this data should be moved up in the hierarchy? If it's under ParallelConsumer interface for example we can use the same api regardless of implementation.

Also, current "requirements" for health-check is pretty simple and used for rolling updates and error-handling:

• know when message-processing starts so that existing instances can be removed (we check the !isClosedOrFailed now, but could perhaps be checking (running || paused))
• know when a failure happens so that instance can be terminated (and potentially restarted) (we check failureCause)

[astubbs]

Are you familiar with micrometer? We're probably just going to use that.

[JorgenRingen]

Yes, we use micrometer for both streams and consumer/producer:

io.micrometer.core.instrument.binder.kafka.KafkaStreamsMetrics
io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics

[astubbs]

Superseded by:

• Feature/health metrics micrometer [WIP] #485

[astubbs]

Should we invert this? Move the construction into the metrics class, instead of the state class? (pass the metrics class the state)

[eddyv]

Closing - Stale / superseded by #485