You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ca...@apache.org on 2024/04/10 07:28:18 UTC
(kafka) branch trunk updated: MINOR: Remove dead code of metric forward-rate (#15686)
This is an automated email from the ASF dual-hosted git repository.
cadonna pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 5c855be0150 MINOR: Remove dead code of metric forward-rate (#15686)
5c855be0150 is described below
commit 5c855be0150f3c7db6a023d909c79f0c5e1fba14
Author: Bruno Cadonna <ca...@apache.org>
AuthorDate: Wed Apr 10 09:28:11 2024 +0200
MINOR: Remove dead code of metric forward-rate (#15686)
Kafka Streams announced the removal of metric forward-rate in
KIP-444 and removed it completely in AK 3.0. However, we forgot
to remove some code for this metric.
This commit removes the code to create the metric forward-rate.
Reviewers: Chia-Ping Tsai <ch...@gmail.com>, Matthias J. Sax <ma...@confluent.io>
---
.../internals/metrics/ProcessorNodeMetrics.java | 54 ----------------------
.../metrics/ProcessorNodeMetricsTest.java | 37 ---------------
2 files changed, 91 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetrics.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetrics.java
index 1a7d914b941..0c2f98cbc1b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetrics.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetrics.java
@@ -29,7 +29,6 @@ import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetric
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.RECORD_E2E_LATENCY_AVG_DESCRIPTION;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.RECORD_E2E_LATENCY_MAX_DESCRIPTION;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.RECORD_E2E_LATENCY_MIN_DESCRIPTION;
-import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.ROLLUP_VALUE;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TASK_LEVEL_GROUP;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.TOTAL_DESCRIPTION;
import static org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl.addAvgAndMinAndMaxToSensor;
@@ -61,12 +60,6 @@ public class ProcessorNodeMetrics {
private static final String PROCESS_RATE_DESCRIPTION =
RATE_DESCRIPTION_PREFIX + PROCESS_DESCRIPTION + RATE_DESCRIPTION_SUFFIX;
- private static final String FORWARD = "forward";
- private static final String FORWARD_DESCRIPTION = "calls to forward";
- private static final String FORWARD_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + FORWARD_DESCRIPTION;
- private static final String FORWARD_RATE_DESCRIPTION =
- RATE_DESCRIPTION_PREFIX + FORWARD_DESCRIPTION + RATE_DESCRIPTION_SUFFIX;
-
private static final String EMITTED_RECORDS = "window-aggregate-final-emit";
private static final String EMITTED_RECORDS_DESCRIPTION = "emit final records";
private static final String EMITTED_RECORDS_TOTAL_DESCRIPTION = TOTAL_DESCRIPTION + EMITTED_RECORDS_DESCRIPTION;
@@ -136,32 +129,6 @@ public class ProcessorNodeMetrics {
);
}
- public static Sensor forwardSensor(final String threadId,
- final String taskId,
- final String processorNodeId,
- final StreamsMetricsImpl streamsMetrics) {
- final Sensor parentSensor = throughputParentSensor(
- threadId,
- taskId,
- FORWARD,
- FORWARD_RATE_DESCRIPTION,
- FORWARD_TOTAL_DESCRIPTION,
- RecordingLevel.DEBUG,
- streamsMetrics
- );
- return throughputSensor(
- threadId,
- taskId,
- processorNodeId,
- FORWARD,
- FORWARD_RATE_DESCRIPTION,
- FORWARD_TOTAL_DESCRIPTION,
- RecordingLevel.DEBUG,
- streamsMetrics,
- parentSensor
- );
- }
-
public static Sensor e2ELatencySensor(final String threadId,
final String taskId,
final String processorNodeId,
@@ -217,27 +184,6 @@ public class ProcessorNodeMetrics {
return sensor;
}
- private static Sensor throughputParentSensor(final String threadId,
- final String taskId,
- final String operation,
- final String descriptionOfRate,
- final String descriptionOfCount,
- final RecordingLevel recordingLevel,
- final StreamsMetricsImpl streamsMetrics) {
- // use operation name as sensor suffix and metric prefix
- final Sensor sensor = streamsMetrics.taskLevelSensor(threadId, taskId, operation, recordingLevel);
- final Map<String, String> parentTagMap = streamsMetrics.nodeLevelTagMap(threadId, taskId, ROLLUP_VALUE);
- addInvocationRateAndCountToSensor(
- sensor,
- PROCESSOR_NODE_LEVEL_GROUP,
- parentTagMap,
- operation,
- descriptionOfRate,
- descriptionOfCount
- );
- return sensor;
- }
-
private static Sensor throughputSensor(final String threadId,
final String taskId,
final String processorNodeId,
diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java
index 738d83683e1..d018affdda2 100644
--- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/metrics/ProcessorNodeMetricsTest.java
@@ -118,43 +118,6 @@ public class ProcessorNodeMetricsTest {
}
}
- @Test
- public void shouldGetForwardSensor() {
- final String metricNamePrefix = "forward";
- final String descriptionOfCount = "The total number of calls to forward";
- final String descriptionOfRate = "The average number of calls to forward per second";
- when(streamsMetrics.taskLevelSensor(THREAD_ID, TASK_ID, metricNamePrefix, RecordingLevel.DEBUG))
- .thenReturn(expectedParentSensor);
- when(streamsMetrics.nodeLevelTagMap(THREAD_ID, TASK_ID, StreamsMetricsImpl.ROLLUP_VALUE))
- .thenReturn(parentTagMap);
- setUpThroughputSensor(metricNamePrefix, RecordingLevel.DEBUG, expectedParentSensor);
-
- try (final MockedStatic<StreamsMetricsImpl> streamsMetricsStaticMock = mockStatic(StreamsMetricsImpl.class)) {
- final Sensor sensor = ProcessorNodeMetrics.forwardSensor(THREAD_ID, TASK_ID, PROCESSOR_NODE_ID, streamsMetrics);
- streamsMetricsStaticMock.verify(
- () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
- expectedSensor,
- PROCESSOR_NODE_LEVEL_GROUP,
- tagMap,
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount
- )
- );
- streamsMetricsStaticMock.verify(
- () -> StreamsMetricsImpl.addInvocationRateAndCountToSensor(
- expectedParentSensor,
- PROCESSOR_NODE_LEVEL_GROUP,
- parentTagMap,
- metricNamePrefix,
- descriptionOfRate,
- descriptionOfCount
- )
- );
- assertThat(sensor, is(expectedSensor));
- }
- }
-
private void setUpThroughputSensor(final String metricNamePrefix,
final RecordingLevel recordingLevel,
final Sensor... parentSensors) {