[
https://issues.apache.org/jira/browse/KAFKA-7261?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16577689#comment-16577689
]
ASF GitHub Bot commented on KAFKA-7261:
---
rajinisivaram closed pull request #5484: KAFKA-7261: Fix request total metric
to count requests instead of bytes
URL: https://github.com/apache/kafka/pull/5484
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Meter.java
b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Meter.java
index 09263cecae8..91d4461d2b5 100644
--- a/clients/src/main/java/org/apache/kafka/common/metrics/stats/Meter.java
+++ b/clients/src/main/java/org/apache/kafka/common/metrics/stats/Meter.java
@@ -61,6 +61,9 @@ public Meter(SampledStat rateStat, MetricName rateMetricName,
MetricName totalMe
* Construct a Meter with provided time unit and provided {@link
SampledStat} stats for Rate
*/
public Meter(TimeUnit unit, SampledStat rateStat, MetricName
rateMetricName, MetricName totalMetricName) {
+if (!(rateStat instanceof SampledTotal) && !(rateStat instanceof
Count)) {
+throw new IllegalArgumentException("Meter is supported only for
SampledTotal and Count");
+}
this.total = new Total();
this.rate = new Rate(unit, rateStat);
this.rateMetricName = rateMetricName;
@@ -77,6 +80,8 @@ public Meter(TimeUnit unit, SampledStat rateStat, MetricName
rateMetricName, Met
@Override
public void record(MetricConfig config, double value, long timeMs) {
rate.record(config, value, timeMs);
-total.record(config, value, timeMs);
+// Total metrics with Count stat should record 1.0 (as recorded in the
count)
+double totalValue = (rate.stat instanceof Count) ? 1.0 : value;
+total.record(config, totalValue, timeMs);
}
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
b/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
index 355233125bc..55354ac8d64 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/NetworkReceive.java
@@ -146,4 +146,12 @@ public ByteBuffer payload() {
return this.buffer;
}
+/**
+ * Returns the total size of the receive including payload and size buffer
+ * for use in metrics. This is consistent with {@link NetworkSend#size()}
+ */
+public int size() {
+return payload().limit() + size.limit();
+}
+
}
diff --git
a/clients/src/main/java/org/apache/kafka/common/network/Selector.java
b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
index 8ca7fff381a..7e32509933e 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Selector.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/Selector.java
@@ -862,7 +862,7 @@ private void addToCompletedReceives() {
private void addToCompletedReceives(KafkaChannel channel,
Deque stagedDeque) {
NetworkReceive networkReceive = stagedDeque.poll();
this.completedReceives.add(networkReceive);
-this.sensors.recordBytesReceived(channel.id(),
networkReceive.payload().limit());
+this.sensors.recordBytesReceived(channel.id(), networkReceive.size());
}
// only for testing
diff --git
a/clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java
b/clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java
index 59bc84e40de..5c75d03b0e6 100644
--- a/clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/metrics/MetricsTest.java
@@ -465,8 +465,12 @@ public void testRateWindowing() throws Exception {
Sensor s = metrics.sensor("test.sensor", cfg);
MetricName rateMetricName = metrics.metricName("test.rate", "grp1");
MetricName totalMetricName = metrics.metricName("test.total", "grp1");
+MetricName countRateMetricName = metrics.metricName("test.count.rate",
"grp1");
+MetricName countTotalMetricName =
metrics.metricName("test.count.total", "grp1");
s.add(new Meter(TimeUnit.SECONDS, rateMetricName, totalMetricName));
-KafkaMetric totalMetric =
metrics.metrics().get(metrics.metricName("test.total", "grp1"));
+s.add(new Meter(TimeUnit.SECONDS, new Count(), countRateMetricName,
countTotalMetricName));
+KafkaMetric totalMetric = metrics.metrics().get(totalMetricName);
+KafkaMetric countTotalMetric =
metrics.metrics().get(countTotalMetricName);
int sum =