You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by da...@apache.org on 2024/05/07 13:21:17 UTC
(kafka) branch trunk updated: KAFKA-16307; Fix coordinator thread idle ratio (#15835)
This is an automated email from the ASF dual-hosted git repository.
dajac 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 21bf715622e KAFKA-16307; Fix coordinator thread idle ratio (#15835)
21bf715622e is described below
commit 21bf715622e9d05984fa8a2a1f9f12d54b76ce41
Author: Jeff Kim <ki...@gmail.com>
AuthorDate: Tue May 7 09:21:09 2024 -0400
KAFKA-16307; Fix coordinator thread idle ratio (#15835)
This PR fixes the thread idle ratio. We take a similar approach to the kafka request handler idle ratio: https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/server/KafkaRequestHandler.scala#L108-L117
Instead of calculating the actual ratio per thread, we record the time each thread stays idle while waiting for a new event, divided by the number of threads as an approximation.
Reviewers: David Jacot <dj...@confluent.io>
---
.../group/metrics/CoordinatorRuntimeMetrics.java | 6 +--
.../metrics/GroupCoordinatorRuntimeMetrics.java | 33 +++++++---------
.../group/runtime/MultiThreadedEventProcessor.java | 23 +++++------
.../GroupCoordinatorRuntimeMetricsTest.java | 22 ++++-------
.../runtime/MultiThreadedEventProcessorTest.java | 46 ++++++++++++----------
5 files changed, 59 insertions(+), 71 deletions(-)
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/CoordinatorRuntimeMetrics.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/CoordinatorRuntimeMetrics.java
index 8978a040007..93ed704279b 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/CoordinatorRuntimeMetrics.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/CoordinatorRuntimeMetrics.java
@@ -54,10 +54,10 @@ public interface CoordinatorRuntimeMetrics extends AutoCloseable {
void recordEventQueueProcessingTime(long durationMs);
/**
- * Record the thread idle ratio.
- * @param ratio The idle ratio.
+ * Record the thread idle time.
+ * @param idleTimeMs The idle time in milliseconds.
*/
- void recordThreadIdleRatio(double ratio);
+ void recordThreadIdleTime(long idleTimeMs);
/**
* Register the event queue size gauge.
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorRuntimeMetrics.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorRuntimeMetrics.java
index 47a7ae473db..c90ee98b923 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorRuntimeMetrics.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorRuntimeMetrics.java
@@ -22,11 +22,12 @@ import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.metrics.stats.Avg;
import org.apache.kafka.common.metrics.stats.Max;
-import org.apache.kafka.common.metrics.stats.Min;
+import org.apache.kafka.common.metrics.stats.Rate;
import org.apache.kafka.coordinator.group.runtime.CoordinatorRuntime.CoordinatorState;
import java.util.Arrays;
import java.util.Objects;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
@@ -77,7 +78,7 @@ public class GroupCoordinatorRuntimeMetrics implements CoordinatorRuntimeMetrics
/**
* The thread idle sensor.
*/
- private final Sensor threadIdleRatioSensor;
+ private final Sensor threadIdleSensor;
public GroupCoordinatorRuntimeMetrics(Metrics metrics) {
this.metrics = Objects.requireNonNull(metrics);
@@ -95,7 +96,7 @@ public class GroupCoordinatorRuntimeMetrics implements CoordinatorRuntimeMetrics
);
this.numPartitionsFailed = kafkaMetricName(
- NUM_PARTITIONS_METRIC_NAME,
+ NUM_PARTITIONS_METRIC_NAME,
"The number of partitions in Failed state.",
"state", "failed"
);
@@ -120,19 +121,13 @@ public class GroupCoordinatorRuntimeMetrics implements CoordinatorRuntimeMetrics
"The average time it took to load the partitions in the last 30 sec."
), new Avg());
- this.threadIdleRatioSensor = metrics.sensor("ThreadIdleRatio");
- this.threadIdleRatioSensor.add(
- metrics.metricName(
- "thread-idle-ratio-min",
- METRICS_GROUP,
- "The minimum thread idle ratio over the last 30 seconds."
- ), new Min());
- this.threadIdleRatioSensor.add(
- metrics.metricName(
- "thread-idle-ratio-avg",
- METRICS_GROUP,
- "The average thread idle ratio over the last 30 seconds."
- ), new Avg());
+ this.threadIdleSensor = metrics.sensor("ThreadIdleRatio");
+ this.threadIdleSensor.add(metrics.metricName(
+ "thread-idle-ratio-avg",
+ METRICS_GROUP,
+ "The fraction of time the threads spent waiting for an event. This is an average across " +
+ "all coordinator event processor threads."),
+ new Rate(TimeUnit.MILLISECONDS));
}
/**
@@ -156,7 +151,7 @@ public class GroupCoordinatorRuntimeMetrics implements CoordinatorRuntimeMetrics
).forEach(metrics::removeMetric);
metrics.removeSensor(partitionLoadSensor.name());
- metrics.removeSensor(threadIdleRatioSensor.name());
+ metrics.removeSensor(threadIdleSensor.name());
}
/**
@@ -208,8 +203,8 @@ public class GroupCoordinatorRuntimeMetrics implements CoordinatorRuntimeMetrics
public void recordEventQueueProcessingTime(long durationMs) { }
@Override
- public void recordThreadIdleRatio(double ratio) {
- threadIdleRatioSensor.record(ratio);
+ public void recordThreadIdleTime(long idleTimeMs) {
+ threadIdleSensor.record(idleTimeMs);
}
@Override
diff --git a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java
index 0aca6a3b79a..20337a46775 100644
--- a/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java
+++ b/group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessor.java
@@ -126,9 +126,16 @@ public class MultiThreadedEventProcessor implements CoordinatorEventProcessor {
private void handleEvents() {
while (!shuttingDown) {
- recordPollStartTime(time.milliseconds());
+ // We use a single meter for aggregate idle percentage for the thread pool.
+ // Since meter is calculated as total_recorded_value / time_window and
+ // time_window is independent of the number of threads, each recorded idle
+ // time should be discounted by # threads.
+
+ long idleStartTimeMs = time.milliseconds();
CoordinatorEvent event = accumulator.take();
- recordPollEndTime(time.milliseconds());
+ long idleEndTimeMs = time.milliseconds();
+ long idleTimeMs = idleEndTimeMs - idleStartTimeMs;
+ metrics.recordThreadIdleTime(idleTimeMs / threads.size());
if (event != null) {
try {
log.debug("Executing event: {}.", event);
@@ -183,18 +190,6 @@ public class MultiThreadedEventProcessor implements CoordinatorEventProcessor {
log.info("Shutdown completed");
}
}
-
- private void recordPollStartTime(long pollStartMs) {
- this.pollStartMs = pollStartMs;
- this.timeSinceLastPollMs = lastPollMs != 0L ? pollStartMs - lastPollMs : 0;
- this.lastPollMs = pollStartMs;
- }
-
- private void recordPollEndTime(long pollEndMs) {
- long pollTimeMs = pollEndMs - pollStartMs;
- double pollIdleRatio = pollTimeMs * 1.0 / (pollTimeMs + timeSinceLastPollMs);
- metrics.recordThreadIdleRatio(pollIdleRatio);
- }
}
/**
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorRuntimeMetricsTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorRuntimeMetricsTest.java
index f3726aafb17..63e4013d6a2 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorRuntimeMetricsTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/metrics/GroupCoordinatorRuntimeMetricsTest.java
@@ -47,7 +47,6 @@ public class GroupCoordinatorRuntimeMetricsTest {
metrics.metricName("event-queue-size", METRICS_GROUP),
metrics.metricName("partition-load-time-max", METRICS_GROUP),
metrics.metricName("partition-load-time-avg", METRICS_GROUP),
- metrics.metricName("thread-idle-ratio-min", METRICS_GROUP),
metrics.metricName("thread-idle-ratio-avg", METRICS_GROUP)
));
@@ -103,24 +102,17 @@ public class GroupCoordinatorRuntimeMetricsTest {
}
@Test
- public void testThreadIdleRatioSensor() {
+ public void testThreadIdleSensor() {
Time time = new MockTime();
Metrics metrics = new Metrics(time);
- try (GroupCoordinatorRuntimeMetrics runtimeMetrics = new GroupCoordinatorRuntimeMetrics(metrics)) {
- IntStream.range(0, 3).forEach(i -> runtimeMetrics.recordThreadIdleRatio(1.0 / (i + 1)));
-
- org.apache.kafka.common.MetricName metricName = metrics.metricName(
- "thread-idle-ratio-avg", METRICS_GROUP);
-
- KafkaMetric metric = metrics.metrics().get(metricName);
- assertEquals((11.0 / 6.0) / 3.0, metric.metricValue()); // (6/6 + 3/6 + 2/6) / 3
+ GroupCoordinatorRuntimeMetrics runtimeMetrics = new GroupCoordinatorRuntimeMetrics(metrics);
+ IntStream.range(0, 3).forEach(i -> runtimeMetrics.recordThreadIdleTime((i + 1) * 1000L));
- metricName = metrics.metricName(
- "thread-idle-ratio-min", METRICS_GROUP);
- metric = metrics.metrics().get(metricName);
- assertEquals(1.0 / 3.0, metric.metricValue());
- }
+ org.apache.kafka.common.MetricName metricName = metrics.metricName(
+ "thread-idle-ratio-avg", METRICS_GROUP);
+ KafkaMetric metric = metrics.metrics().get(metricName);
+ assertEquals(6 / 30.0, metric.metricValue()); // 'total_ms / window_ms'
}
@Test
diff --git a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java
index f01fc883105..4a1656d5100 100644
--- a/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java
+++ b/group-coordinator/src/test/java/org/apache/kafka/coordinator/group/runtime/MultiThreadedEventProcessorTest.java
@@ -42,7 +42,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.anyDouble;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -73,7 +73,7 @@ public class MultiThreadedEventProcessorTest {
private final boolean block;
private final CountDownLatch latch;
private final CountDownLatch executed;
- private long createdTimeMs;
+ private final long createdTimeMs;
FutureEvent(
TopicPartition key,
@@ -442,8 +442,6 @@ public class MultiThreadedEventProcessorTest {
// e2 poll time = 500
// e2 processing time = 5000
- // e1 poll time / e1 poll time
- verify(mockRuntimeMetrics, times(1)).recordThreadIdleRatio(1.0);
// e1 poll time
verify(mockRuntimeMetrics, times(1)).recordEventQueueTime(500L);
// e1 processing time + e2 enqueue time
@@ -451,8 +449,8 @@ public class MultiThreadedEventProcessorTest {
// Second event (e2)
- // idle ratio = e2 poll time / (e1 poll time + e1 processing time + e2 enqueue time + e2 poll time)
- verify(mockRuntimeMetrics, times(1)).recordThreadIdleRatio(500.0 / (500.0 + 7000.0 + 500.0));
+ // e1, e2 poll time
+ verify(mockRuntimeMetrics, times(2)).recordThreadIdleTime(500L);
// event queue time = e2 enqueue time + e2 poll time
verify(mockRuntimeMetrics, times(1)).recordEventQueueTime(3500L);
}
@@ -461,26 +459,30 @@ public class MultiThreadedEventProcessorTest {
@Test
public void testRecordThreadIdleRatioTwoThreads() throws Exception {
GroupCoordinatorRuntimeMetrics mockRuntimeMetrics = mock(GroupCoordinatorRuntimeMetrics.class);
+ Time time = Time.SYSTEM;
try (CoordinatorEventProcessor eventProcessor = new MultiThreadedEventProcessor(
new LogContext(),
"event-processor-",
2,
- Time.SYSTEM,
+ time,
mockRuntimeMetrics,
- new DelayEventAccumulator(Time.SYSTEM, 100L)
+ new DelayEventAccumulator(time, 100L)
)) {
- List<Double> recordedRatios = new ArrayList<>();
+ List<Long> recordedIdleTimesMs = new ArrayList<>();
AtomicInteger numEventsExecuted = new AtomicInteger(0);
- ArgumentCaptor<Double> ratioCaptured = ArgumentCaptor.forClass(Double.class);
+ ArgumentCaptor<Long> idleTimeCaptured = ArgumentCaptor.forClass(Long.class);
doAnswer(invocation -> {
- double threadIdleRatio = ratioCaptured.getValue();
- assertTrue(threadIdleRatio > 0.0);
- synchronized (recordedRatios) {
- recordedRatios.add(threadIdleRatio);
+ long threadIdleTime = idleTimeCaptured.getValue();
+
+ // As each thread sleeps for 100ms before returning from take(),
+ // we know that each recorded idle time must be at least 50 (100/2)ms.
+ assertTrue(threadIdleTime >= 50);
+ synchronized (recordedIdleTimesMs) {
+ recordedIdleTimesMs.add(threadIdleTime);
}
return null;
- }).when(mockRuntimeMetrics).recordThreadIdleRatio(ratioCaptured.capture());
+ }).when(mockRuntimeMetrics).recordThreadIdleTime(idleTimeCaptured.capture());
List<FutureEvent<Integer>> events = Arrays.asList(
new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet),
@@ -489,9 +491,11 @@ public class MultiThreadedEventProcessorTest {
new FutureEvent<>(new TopicPartition("foo", 0), numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 1), numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 2), numEventsExecuted::incrementAndGet),
+ new FutureEvent<>(new TopicPartition("foo", 2), numEventsExecuted::incrementAndGet),
new FutureEvent<>(new TopicPartition("foo", 2), numEventsExecuted::incrementAndGet)
);
+ long startMs = time.milliseconds();
events.forEach(eventProcessor::enqueueLast);
CompletableFuture.allOf(events
@@ -499,18 +503,20 @@ public class MultiThreadedEventProcessorTest {
.map(FutureEvent::future)
.toArray(CompletableFuture[]::new)
).get(10, TimeUnit.SECONDS);
-
events.forEach(event -> {
assertTrue(event.future.isDone());
assertFalse(event.future.isCompletedExceptionally());
});
+ long diff = time.milliseconds() - startMs;
+
assertEquals(events.size(), numEventsExecuted.get());
- verify(mockRuntimeMetrics, times(7)).recordThreadIdleRatio(anyDouble());
+ verify(mockRuntimeMetrics, times(8)).recordThreadIdleTime(anyLong());
+ assertEquals(8, recordedIdleTimesMs.size());
- assertEquals(7, recordedRatios.size());
- double average = recordedRatios.stream().mapToDouble(Double::doubleValue).sum() / 7;
- assertTrue(average > 0.0 && average < 1.0);
+ long sum = recordedIdleTimesMs.stream().mapToLong(Long::longValue).sum();
+ double idleRatio = (double) sum / diff;
+ assertTrue(idleRatio >= 0.5 && idleRatio <= 1.0);
}
}
}