This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.9 by this push:
new 85658e5e33f KAFKA-18597 Fix max-buffer-utilization-percent is always 0
(#18690)
85658e5e33f is described below
commit 85658e5e33f550c6953ae58e37cc1808fd56b879
Author: Logan Zhu <[email protected]>
AuthorDate: Tue Jan 28 02:23:59 2025 +0800
KAFKA-18597 Fix max-buffer-utilization-percent is always 0 (#18690)
Reviewers: Chia-Ping Tsai <[email protected]>
---
core/src/main/scala/kafka/log/LogCleaner.scala | 20 +--
.../test/scala/unit/kafka/log/LogCleanerTest.scala | 159 ++++++++++++++++++++-
2 files changed, 168 insertions(+), 11 deletions(-)
diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala
b/core/src/main/scala/kafka/log/LogCleaner.scala
index 1265e979373..96bcf7c86ed 100644
--- a/core/src/main/scala/kafka/log/LogCleaner.scala
+++ b/core/src/main/scala/kafka/log/LogCleaner.scala
@@ -116,16 +116,15 @@ class LogCleaner(initialConfig: CleanerConfig,
private[log] val cleaners = mutable.ArrayBuffer[CleanerThread]()
/**
- * scala 2.12 does not support maxOption so we handle the empty manually.
* @param f to compute the result
- * @return the max value (int value) or 0 if there is no cleaner
+ * @return the max value or 0 if there is no cleaner
*/
- private def maxOverCleanerThreads(f: CleanerThread => Double): Int =
- cleaners.foldLeft(0.0d)((max: Double, thread: CleanerThread) =>
math.max(max, f(thread))).toInt
+ private[log] def maxOverCleanerThreads(f: CleanerThread => Double): Double =
+ cleaners.foldLeft(0.0d)((max: Double, thread: CleanerThread) =>
math.max(max, f(thread)))
/* a metric to track the maximum utilization of any thread's buffer in the
last cleaning */
metricsGroup.newGauge(MaxBufferUtilizationPercentMetricName,
- () => maxOverCleanerThreads(_.lastStats.bufferUtilization) * 100)
+ () => (maxOverCleanerThreads(_.lastStats.bufferUtilization) * 100).toInt)
/* a metric to track the recopy rate of each thread's last cleaning */
metricsGroup.newGauge(CleanerRecopyPercentMetricName, () => {
@@ -135,12 +134,12 @@ class LogCleaner(initialConfig: CleanerConfig,
})
/* a metric to track the maximum cleaning time for the last cleaning from
each thread */
- metricsGroup.newGauge(MaxCleanTimeMetricName, () =>
maxOverCleanerThreads(_.lastStats.elapsedSecs))
+ metricsGroup.newGauge(MaxCleanTimeMetricName, () =>
maxOverCleanerThreads(_.lastStats.elapsedSecs).toInt)
// a metric to track delay between the time when a log is required to be
compacted
// as determined by max compaction lag and the time of last cleaner run.
metricsGroup.newGauge(MaxCompactionDelayMetricsName,
- () =>
maxOverCleanerThreads(_.lastPreCleanStats.maxCompactionDelayMs.toDouble) / 1000)
+ () =>
(maxOverCleanerThreads(_.lastPreCleanStats.maxCompactionDelayMs.toDouble) /
1000).toInt)
metricsGroup.newGauge(DeadThreadCountMetricName, () => deadThreadCount)
@@ -524,10 +523,11 @@ object LogCleaner {
}
- private val MaxBufferUtilizationPercentMetricName =
"max-buffer-utilization-percent"
+ // Visible for test.
+ private[log] val MaxBufferUtilizationPercentMetricName =
"max-buffer-utilization-percent"
private val CleanerRecopyPercentMetricName = "cleaner-recopy-percent"
- private val MaxCleanTimeMetricName = "max-clean-time-secs"
- private val MaxCompactionDelayMetricsName = "max-compaction-delay-secs"
+ private[log] val MaxCleanTimeMetricName = "max-clean-time-secs"
+ private[log] val MaxCompactionDelayMetricsName = "max-compaction-delay-secs"
private val DeadThreadCountMetricName = "DeadThreadCount"
// package private for testing
private[log] val MetricNames = Set(
diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
index ee62b3f0f21..cb22931bb88 100644
--- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala
@@ -17,7 +17,8 @@
package kafka.log
-import kafka.common._
+import kafka.common.LogCleaningAbortedException
+import kafka.log.LogCleaner.{MaxBufferUtilizationPercentMetricName,
MaxCleanTimeMetricName, MaxCompactionDelayMetricsName}
import kafka.server.{BrokerTopicStats, KafkaConfig}
import kafka.utils.{CoreUtils, Logging, Pool, TestUtils}
import org.apache.kafka.common.TopicPartition
@@ -2009,6 +2010,162 @@ class LogCleanerTest extends Logging {
}
}
+ @Test
+ def testMaxBufferUtilizationPercentMetric(): Unit = {
+ val logCleaner = new LogCleaner(
+ new CleanerConfig(true),
+ logDirs = Array(TestUtils.tempDir(), TestUtils.tempDir()),
+ logs = new Pool[TopicPartition, UnifiedLog](),
+ logDirFailureChannel = new LogDirFailureChannel(1),
+ time = time
+ )
+
+ def assertMaxBufferUtilizationPercent(expected: Int): Unit = {
+ val gauge =
logCleaner.metricsGroup.newGauge(MaxBufferUtilizationPercentMetricName,
+ () => (logCleaner.maxOverCleanerThreads(_.lastStats.bufferUtilization)
* 100).toInt)
+ assertEquals(expected, gauge.value())
+ }
+
+ try {
+ // No CleanerThreads
+ assertMaxBufferUtilizationPercent(0)
+
+ val cleaners = logCleaner.cleaners
+
+ val cleaner1 = new logCleaner.CleanerThread(1)
+ cleaner1.lastStats = new CleanerStats(time)
+ cleaner1.lastStats.bufferUtilization = 0.75
+ cleaners += cleaner1
+
+ val cleaner2 = new logCleaner.CleanerThread(2)
+ cleaner2.lastStats = new CleanerStats(time)
+ cleaner2.lastStats.bufferUtilization = 0.85
+ cleaners += cleaner2
+
+ val cleaner3 = new logCleaner.CleanerThread(3)
+ cleaner3.lastStats = new CleanerStats(time)
+ cleaner3.lastStats.bufferUtilization = 0.65
+ cleaners += cleaner3
+
+ // expect the gauge value to reflect the maximum bufferUtilization
+ assertMaxBufferUtilizationPercent(85)
+
+ // Update bufferUtilization and verify the gauge value updates
+ cleaner1.lastStats.bufferUtilization = 0.9
+ assertMaxBufferUtilizationPercent(90)
+
+ // All CleanerThreads have the same bufferUtilization
+ cleaners.foreach(_.lastStats.bufferUtilization = 0.5)
+ assertMaxBufferUtilizationPercent(50)
+ } finally {
+ logCleaner.shutdown()
+ }
+ }
+
+ @Test
+ def testMaxCleanTimeMetric(): Unit = {
+ val logCleaner = new LogCleaner(
+ new CleanerConfig(true),
+ logDirs = Array(TestUtils.tempDir(), TestUtils.tempDir()),
+ logs = new Pool[TopicPartition, UnifiedLog](),
+ logDirFailureChannel = new LogDirFailureChannel(1),
+ time = time
+ )
+
+ def assertMaxCleanTime(expected: Int): Unit = {
+ val gauge = logCleaner.metricsGroup.newGauge(MaxCleanTimeMetricName,
+ () => logCleaner.maxOverCleanerThreads(_.lastStats.elapsedSecs).toInt)
+ assertEquals(expected, gauge.value())
+ }
+
+ try {
+ // No CleanerThreads
+ assertMaxCleanTime(0)
+
+ val cleaners = logCleaner.cleaners
+
+ val cleaner1 = new logCleaner.CleanerThread(1)
+ cleaner1.lastStats = new CleanerStats(time)
+ cleaner1.lastStats.endTime = cleaner1.lastStats.startTime + 1000L
+ cleaners += cleaner1
+
+ val cleaner2 = new logCleaner.CleanerThread(2)
+ cleaner2.lastStats = new CleanerStats(time)
+ cleaner2.lastStats.endTime = cleaner2.lastStats.startTime + 2000L
+ cleaners += cleaner2
+
+ val cleaner3 = new logCleaner.CleanerThread(3)
+ cleaner3.lastStats = new CleanerStats(time)
+ cleaner3.lastStats.endTime = cleaner3.lastStats.startTime + 3000L
+ cleaners += cleaner3
+
+ // expect the gauge value to reflect the maximum cleanTime
+ assertMaxCleanTime(3)
+
+ // Update cleanTime and verify the gauge value updates
+ cleaner1.lastStats.endTime = cleaner1.lastStats.startTime + 4000L
+ assertMaxCleanTime(4)
+
+ // All CleanerThreads have the same cleanTime
+ cleaners.foreach(cleaner => cleaner.lastStats.endTime =
cleaner.lastStats.startTime + 1500L)
+ assertMaxCleanTime(1)
+ } finally {
+ logCleaner.shutdown()
+ }
+ }
+
+ @Test
+ def testMaxCompactionDelayMetrics(): Unit = {
+ val logCleaner = new LogCleaner(
+ new CleanerConfig(true),
+ logDirs = Array(TestUtils.tempDir(), TestUtils.tempDir()),
+ logs = new Pool[TopicPartition, UnifiedLog](),
+ logDirFailureChannel = new LogDirFailureChannel(1),
+ time = time
+ )
+
+ def assertMaxCompactionDelay(expected: Int): Unit = {
+ val gauge =
logCleaner.metricsGroup.newGauge(MaxCompactionDelayMetricsName,
+ () =>
(logCleaner.maxOverCleanerThreads(_.lastPreCleanStats.maxCompactionDelayMs.toDouble)
/ 1000).toInt)
+ assertEquals(expected, gauge.value())
+ }
+
+ try {
+ // No CleanerThreads
+ assertMaxCompactionDelay(0)
+
+ val cleaners = logCleaner.cleaners
+
+ val cleaner1 = new logCleaner.CleanerThread(1)
+ cleaner1.lastStats = new CleanerStats(time)
+ cleaner1.lastPreCleanStats.maxCompactionDelayMs = 1000L
+ cleaners += cleaner1
+
+ val cleaner2 = new logCleaner.CleanerThread(2)
+ cleaner2.lastStats = new CleanerStats(time)
+ cleaner2.lastPreCleanStats.maxCompactionDelayMs = 2000L
+ cleaners += cleaner2
+
+ val cleaner3 = new logCleaner.CleanerThread(3)
+ cleaner3.lastStats = new CleanerStats(time)
+ cleaner3.lastPreCleanStats.maxCompactionDelayMs = 3000L
+ cleaners += cleaner3
+
+ // expect the gauge value to reflect the maximum CompactionDelay
+ assertMaxCompactionDelay(3)
+
+ // Update CompactionDelay and verify the gauge value updates
+ cleaner1.lastPreCleanStats.maxCompactionDelayMs = 4000L
+ assertMaxCompactionDelay(4)
+
+ // All CleanerThreads have the same CompactionDelay
+ cleaners.foreach(_.lastPreCleanStats.maxCompactionDelayMs = 1500L)
+ assertMaxCompactionDelay(1)
+ } finally {
+ logCleaner.shutdown()
+ }
+ }
+
private def writeToLog(log: UnifiedLog, keysAndValues: Iterable[(Int, Int)],
offsetSeq: Iterable[Long]): Iterable[Long] = {
for (((key, value), offset) <- keysAndValues.zip(offsetSeq))
yield log.appendAsFollower(messageWithOffset(key, value,
offset)).lastOffset
