Metrics in Kafka Source

[kkondaka]

Is your feature request related to a problem? Please describe.
DataPrepper recently added Kafka Source support. Kafka source should expose KafkaConsumer metrics and other metrics to DataPrepper users using the PluginMetrics

Describe the solution you'd like
Types of metrics to support

• global (KafkaSource level) metrics
• topic level metrics
  • KafkaConsumer metrics (obtained by using metrics() API of KafkaConsumer
  • Internal metrics that track the functioning of KafkaConsumer code. Examples of these metrics are - number of positive/negative acknowledgements received, number of auth exceptions, etc

Since Kafka Source allows creation of multiple threads per topic, we need a way to efficiently collect and expose metrics.

Proposal

• Each thread maintains local counters and exports them periodically to a common/shared data structure (shared by all threads consuming from the same topic)
• Periodically, per-topic metrics are calculated by aggregating the metrics from each thread and PluginMetric.guage() is created. Per topic metrics will have topic. prefix followed by topicname followed by . before the metricName. Also Kafka metric names with - will be replaced by camel case equivalent. For example, the KafkaConsumer metric records-consumed per topic will be topic.topicName.recordsConsumed
• Global metrics are aggregated across all threads (irrespective of topic)

Describe alternatives you've considered (Optional)
Alternative is to use synchronized blocks or atomic counters to increment the metrics inline. I think this method will be very inefficient when the number of topics or threads are high.

Additional context
Add any other context or screenshots about the feature request here.

[dlvenable]

I agree that using . between these parts of the metric names makes sense. It could easily feed into different dimensions. I also agree with replacing the - to keep the metric names consistent with other Data Prepper names.