You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by je...@apache.org on 2025/01/21 19:33:31 UTC
(kafka) branch kafka-18610-1 updated (b1cc483c4ba -> e54e2f5d160)
This is an automated email from the ASF dual-hosted git repository.
jeffkbkim pushed a change to branch kafka-18610-1
in repository https://gitbox.apache.org/repos/asf/kafka.git
discard b1cc483c4ba add ceiling to all histogram recorded values
discard 75a55e228a5 add batch flush interval time ms metric
new e54e2f5d160 add batch flush interval time ms metric
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (b1cc483c4ba)
\
N -- N -- N refs/heads/kafka-18610-1 (e54e2f5d160)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
(kafka) 01/01: add batch flush interval time ms metric
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jeffkbkim pushed a commit to branch kafka-18610-1
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit e54e2f5d160bbb50cb46b4cf01cc54bf25025c69
Author: Jeff Kim <ki...@gmail.com>
AuthorDate: Tue Dec 17 16:15:27 2024 -0500
add batch flush interval time ms metric
---
.../common/runtime/CoordinatorRuntime.java | 1 +
.../common/runtime/CoordinatorRuntimeMetrics.java | 7 ++++++
.../runtime/CoordinatorRuntimeMetricsImpl.java | 25 ++++++++++++++++++++++
.../common/runtime/KafkaMetricHistogram.java | 4 +++-
.../runtime/CoordinatorRuntimeMetricsImplTest.java | 14 ++++++++++--
5 files changed, 48 insertions(+), 3 deletions(-)
diff --git a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
index 5d9e9ef13a8..1da472f615c 100644
--- a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
+++ b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java
@@ -780,6 +780,7 @@ public class CoordinatorRuntime<S extends CoordinatorShard<U>, U> implements Aut
}
long flushStartMs = time.milliseconds();
+ runtimeMetrics.recordFlushIntervalTime(flushStartMs - currentBatch.appendTimeMs);
// Write the records to the log and update the last written offset.
long offset = partitionWriter.append(
tp,
diff --git a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetrics.java b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetrics.java
index 0218fa13fa8..6c876931ab0 100644
--- a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetrics.java
+++ b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetrics.java
@@ -67,6 +67,13 @@ public interface CoordinatorRuntimeMetrics extends AutoCloseable {
*/
void recordFlushTime(long durationMs);
+ /**
+ * Record the flush time.
+ *
+ * @param durationMs The flush time in milliseconds.
+ */
+ void recordFlushIntervalTime(long durationMs);
+
/**
* Record the thread idle time.
* @param idleTimeMs The idle time in milliseconds.
diff --git a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java
index 391813250c1..c59521a04b0 100644
--- a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java
+++ b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImpl.java
@@ -65,6 +65,11 @@ public class CoordinatorRuntimeMetricsImpl implements CoordinatorRuntimeMetrics
*/
public static final String BATCH_FLUSH_TIME_METRIC_NAME = "batch-flush-time-ms";
+ /**
+ * The flush time metric name.
+ */
+ public static final String BATCH_FLUSH_INTERVAL_METRIC_NAME = "batch-flush-interval-time-ms";
+
/**
* Metric to count the number of partitions in Loading state.
*/
@@ -123,6 +128,11 @@ public class CoordinatorRuntimeMetricsImpl implements CoordinatorRuntimeMetrics
*/
private final Sensor flushTimeSensor;
+ /**
+ * Sensor to measure the flush interval time.
+ */
+ private final Sensor flushIntervalTimeSensor;
+
public CoordinatorRuntimeMetricsImpl(Metrics metrics, String metricsGroup) {
this.metrics = Objects.requireNonNull(metrics);
this.metricsGroup = Objects.requireNonNull(metricsGroup);
@@ -209,6 +219,15 @@ public class CoordinatorRuntimeMetricsImpl implements CoordinatorRuntimeMetrics
);
this.flushTimeSensor = metrics.sensor("FlushTime");
this.flushTimeSensor.add(flushTimeHistogram);
+
+ KafkaMetricHistogram flushIntervalHistogram = KafkaMetricHistogram.newLatencyHistogram(
+ suffix -> kafkaMetricName(
+ BATCH_FLUSH_INTERVAL_METRIC_NAME + "-" + suffix,
+ "The " + suffix + " flush interval time in milliseconds"
+ )
+ );
+ this.flushIntervalTimeSensor = metrics.sensor("FlushIntervalTime");
+ this.flushIntervalTimeSensor.add(flushIntervalHistogram);
}
/**
@@ -237,6 +256,7 @@ public class CoordinatorRuntimeMetricsImpl implements CoordinatorRuntimeMetrics
metrics.removeSensor(eventProcessingTimeSensor.name());
metrics.removeSensor(eventPurgatoryTimeSensor.name());
metrics.removeSensor(flushTimeSensor.name());
+ metrics.removeSensor(flushIntervalTimeSensor.name());
}
/**
@@ -301,6 +321,11 @@ public class CoordinatorRuntimeMetricsImpl implements CoordinatorRuntimeMetrics
flushTimeSensor.record(durationMs);
}
+ @Override
+ public void recordFlushIntervalTime(long durationMs) {
+ flushIntervalTimeSensor.record(durationMs);
+ }
+
@Override
public void recordThreadIdleTime(long idleTimeMs) {
threadIdleSensor.record(idleTimeMs);
diff --git a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/KafkaMetricHistogram.java b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/KafkaMetricHistogram.java
index 9f9575b5d98..a913bc8697a 100644
--- a/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/KafkaMetricHistogram.java
+++ b/coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/KafkaMetricHistogram.java
@@ -67,6 +67,7 @@ public final class KafkaMetricHistogram implements CompoundStat {
*/
private final Function<String, MetricName> metricNameFactory;
private final HdrHistogram hdrHistogram;
+ private final long highestTrackableValue;
/**
* Creates a new histogram with the purpose of tracking latency values. As such, the histogram
@@ -105,6 +106,7 @@ public final class KafkaMetricHistogram implements CompoundStat {
) {
this.metricNameFactory = metricNameFactory;
this.hdrHistogram = new HdrHistogram(highestTrackableValue, numberOfSignificantValueDigits);
+ this.highestTrackableValue = highestTrackableValue;
}
@Override
@@ -125,6 +127,6 @@ public final class KafkaMetricHistogram implements CompoundStat {
@Override
public void record(MetricConfig config, double value, long timeMs) {
- hdrHistogram.record((long) value);
+ hdrHistogram.record(Math.min(highestTrackableValue, (long) value));
}
}
diff --git a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImplTest.java b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImplTest.java
index 7285b58ffab..9452486ce7a 100644
--- a/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImplTest.java
+++ b/coordinator-common/src/test/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntimeMetricsImplTest.java
@@ -31,6 +31,7 @@ import java.util.Arrays;
import java.util.HashSet;
import java.util.stream.IntStream;
+import static org.apache.kafka.coordinator.common.runtime.CoordinatorRuntimeMetricsImpl.BATCH_FLUSH_INTERVAL_METRIC_NAME;
import static org.apache.kafka.coordinator.common.runtime.CoordinatorRuntimeMetricsImpl.BATCH_FLUSH_TIME_METRIC_NAME;
import static org.apache.kafka.coordinator.common.runtime.CoordinatorRuntimeMetricsImpl.EVENT_PROCESSING_TIME_METRIC_NAME;
import static org.apache.kafka.coordinator.common.runtime.CoordinatorRuntimeMetricsImpl.EVENT_PURGATORY_TIME_METRIC_NAME;
@@ -76,7 +77,12 @@ public class CoordinatorRuntimeMetricsImplTest {
kafkaMetricName(metrics, "batch-flush-time-ms-p50"),
kafkaMetricName(metrics, "batch-flush-time-ms-p95"),
kafkaMetricName(metrics, "batch-flush-time-ms-p99"),
- kafkaMetricName(metrics, "batch-flush-time-ms-p999")
+ kafkaMetricName(metrics, "batch-flush-time-ms-p999"),
+ kafkaMetricName(metrics, "batch-flush-interval-time-ms-max"),
+ kafkaMetricName(metrics, "batch-flush-interval-time-ms-p50"),
+ kafkaMetricName(metrics, "batch-flush-interval-time-ms-p95"),
+ kafkaMetricName(metrics, "batch-flush-interval-time-ms-p99"),
+ kafkaMetricName(metrics, "batch-flush-interval-time-ms-p999")
));
try (CoordinatorRuntimeMetricsImpl runtimeMetrics = new CoordinatorRuntimeMetricsImpl(metrics, METRICS_GROUP)) {
@@ -160,7 +166,8 @@ public class CoordinatorRuntimeMetricsImplTest {
EVENT_QUEUE_TIME_METRIC_NAME,
EVENT_PROCESSING_TIME_METRIC_NAME,
EVENT_PURGATORY_TIME_METRIC_NAME,
- BATCH_FLUSH_TIME_METRIC_NAME
+ BATCH_FLUSH_TIME_METRIC_NAME,
+ BATCH_FLUSH_INTERVAL_METRIC_NAME
})
public void testHistogramMetrics(String metricNamePrefix) {
Time time = new MockTime();
@@ -181,6 +188,9 @@ public class CoordinatorRuntimeMetricsImplTest {
break;
case BATCH_FLUSH_TIME_METRIC_NAME:
runtimeMetrics.recordFlushTime(i);
+ break;
+ case BATCH_FLUSH_INTERVAL_METRIC_NAME:
+ runtimeMetrics.recordFlushIntervalTime(i);
}
});