stevenschlansker commented on code in PR #23763:
URL: https://github.com/apache/kafka/pull/23763#discussion_r4483839057
##########
clients/src/main/java/org/apache/kafka/common/telemetry/internals/KafkaMetricsCollector.java:
##########
@@ -179,6 +179,10 @@ public void collect(MetricsEmitter metricsEmitter) {
KafkaMetric metric = entry.getValue();
try {
+ if (!metric.isActive()) {
+ log.warn("Skipping inactive kafkaMetric {}",
metricKey.name());
Review Comment:
I think this warning could be too chatty - if my read is right, this will
fire on every emit, which for us would log ~1000 warnings per interval in our
case
##########
clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java:
##########
@@ -83,6 +105,16 @@ public Object metricValue() {
}
}
+ /**
+ * The method determines if the metric is active based on the recording
level and the metric's configuration.
+ * @return true if the metric is active, false otherwise
+ */
+ public boolean isActive() {
Review Comment:
Same concern here: making this part of public API might trigger maintainers
asking for a KIP, so consider a non-public accessor (maybe needing an internal
helper class to avoid Java level access issues)
##########
clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java:
##########
@@ -44,12 +45,33 @@ public final class KafkaMetric implements Metric {
*/
@SuppressKafkaInternalApiUsage("KIP-1265: ctor leaks internal Time for
test injection — pending KIP review to promote Time or refactor")
public KafkaMetric(Object lock, MetricName metricName,
MetricValueProvider<?> valueProvider,
- MetricConfig config, Time time) {
+ MetricConfig config, Time time) {
this.metricName = metricName;
this.lock = lock;
this.metricValueProvider = Objects.requireNonNull(valueProvider,
"valueProvider must not be null");
this.config = config;
this.time = time;
+ this.recordingLevel = Optional.empty();
+ }
+
+ /**
+ * Constructor for KafkaMetric having recording level.
+ * @param lock The lock used to prevent race condition
+ * @param metricName The name of the metric
+ * @param valueProvider The metric value provider associated with this
metric
+ * @param config The configuration of the metric
+ * @param time The time instance to use with the metrics
+ * @param recordingLevel The recording level of the metric
+ */
+
+ public KafkaMetric(Object lock, MetricName metricName,
MetricValueProvider<?> valueProvider,
Review Comment:
I am not an expert on Kafka policies, but since this constructor is used
internally, we could make it non-public to avoid maintainers asking for a KIP.
##########
clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java:
##########
@@ -44,12 +45,33 @@ public final class KafkaMetric implements Metric {
*/
@SuppressKafkaInternalApiUsage("KIP-1265: ctor leaks internal Time for
test injection — pending KIP review to promote Time or refactor")
public KafkaMetric(Object lock, MetricName metricName,
MetricValueProvider<?> valueProvider,
- MetricConfig config, Time time) {
+ MetricConfig config, Time time) {
this.metricName = metricName;
this.lock = lock;
this.metricValueProvider = Objects.requireNonNull(valueProvider,
"valueProvider must not be null");
this.config = config;
this.time = time;
+ this.recordingLevel = Optional.empty();
+ }
+
+ /**
+ * Constructor for KafkaMetric having recording level.
+ * @param lock The lock used to prevent race condition
+ * @param metricName The name of the metric
+ * @param valueProvider The metric value provider associated with this
metric
+ * @param config The configuration of the metric
+ * @param time The time instance to use with the metrics
+ * @param recordingLevel The recording level of the metric
+ */
+
+ public KafkaMetric(Object lock, MetricName metricName,
MetricValueProvider<?> valueProvider,
+ MetricConfig config, Time time, Sensor.RecordingLevel
recordingLevel) {
+ this.metricName = metricName;
+ this.lock = lock;
+ this.metricValueProvider = Objects.requireNonNull(valueProvider,
"valueProvider must not be null");
Review Comment:
Can this constructor body be DRY by making one constructor delegate to the
other?
##########
clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java:
##########
@@ -44,12 +45,33 @@ public final class KafkaMetric implements Metric {
*/
@SuppressKafkaInternalApiUsage("KIP-1265: ctor leaks internal Time for
test injection — pending KIP review to promote Time or refactor")
public KafkaMetric(Object lock, MetricName metricName,
MetricValueProvider<?> valueProvider,
- MetricConfig config, Time time) {
+ MetricConfig config, Time time) {
this.metricName = metricName;
this.lock = lock;
this.metricValueProvider = Objects.requireNonNull(valueProvider,
"valueProvider must not be null");
this.config = config;
this.time = time;
+ this.recordingLevel = Optional.empty();
+ }
+
+ /**
+ * Constructor for KafkaMetric having recording level.
+ * @param lock The lock used to prevent race condition
+ * @param metricName The name of the metric
+ * @param valueProvider The metric value provider associated with this
metric
+ * @param config The configuration of the metric
+ * @param time The time instance to use with the metrics
+ * @param recordingLevel The recording level of the metric
+ */
+
Review Comment:
nit: stray blank line
##########
clients/src/main/java/org/apache/kafka/common/metrics/KafkaMetric.java:
##########
@@ -83,6 +105,16 @@ public Object metricValue() {
}
}
+ /**
+ * The method determines if the metric is active based on the recording
level and the metric's configuration.
+ * @return true if the metric is active, false otherwise
+ */
+ public boolean isActive() {
+ return recordingLevel
+ .map(level -> level.shouldRecord(config.recordLevel().id))
Review Comment:
Claude finding:
```
Check the level against the same config the sensor uses.
Sensor.shouldRecord() uses the sensor's this.config.recordLevel(). isActive()
uses
statConfig, which is the per-stat MetricConfig when a caller passes one
to Sensor.add(..., config). A fresh MetricConfig defaults to INFO.
- Failure case: a DEBUG sensor adds a stat with new
MetricConfig().samples(5). The application runs at DEBUG and the sensor
records. The metric
still never reaches telemetry.
- Simplest fix: give KafkaMetric a BooleanSupplier that Sensor.add sets
to this::shouldRecord, with () -> true for metrics that no sensor owns.
The telemetry filter then matches recording exactly, and the
Optional<RecordingLevel> goes away.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
