MINOR: log new coordinator partition load schedule time (apache#15017)

The current load summary exposes the time from when the partition load operation is scheduled to when the load completes. We are missing the information of how long the scheduled operation stays in the scheduler. Log that information. Reviewers: David Jacot <[email protected]>
tarunsankhla · Jan 18, 2024 · 96f852f · 96f852f
1 parent acab465
commit 96f852f
Show file tree

Hide file tree

Showing 4 changed files with 18 additions and 5 deletions.
diff --git a/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala b/core/src/main/scala/kafka/coordinator/group/CoordinatorLoaderImpl.scala
@@ -79,6 +79,7 @@ class CoordinatorLoaderImpl[T](
     future: CompletableFuture[LoadSummary],
     startTimeMs: Long
   ): Unit = {
+    val schedulerQueueTimeMs = time.milliseconds() - startTimeMs
     try {
       replicaManager.getLog(tp) match {
         case None =>
@@ -193,7 +194,7 @@ class CoordinatorLoaderImpl[T](
               s"Stopped loading records from $tp because the partition is not online or is no longer the leader."
             ))
           } else if (isRunning.get) {
-            future.complete(new LoadSummary(startTimeMs, endTimeMs, numRecords, numBytes))
+            future.complete(new LoadSummary(startTimeMs, endTimeMs, schedulerQueueTimeMs, numRecords, numBytes))
           } else {
             future.completeExceptionally(new RuntimeException("Coordinator loader is closed."))
           }

diff --git a/...ordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorLoader.java b/...ordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorLoader.java
@@ -53,12 +53,14 @@ public short unknownType() {
     class LoadSummary {
         private final long startTimeMs;
         private final long endTimeMs;
+        private final long schedulerQueueTimeMs;
         private final long numRecords;
         private final long numBytes;
 
-        public LoadSummary(long startTimeMs, long endTimeMs, long numRecords, long numBytes) {
+        public LoadSummary(long startTimeMs, long endTimeMs, long schedulerQueueTimeMs, long numRecords, long numBytes) {
             this.startTimeMs = startTimeMs;
             this.endTimeMs = endTimeMs;
+            this.schedulerQueueTimeMs = schedulerQueueTimeMs;
             this.numRecords = numRecords;
             this.numBytes = numBytes;
         }
@@ -71,6 +73,10 @@ public long endTimeMs() {
             return endTimeMs;
         }
 
+        public long schedulerQueueTimeMs() {
+            return schedulerQueueTimeMs;
+        }
+
         public long numRecords() {
             return numRecords;
         }
@@ -84,6 +90,7 @@ public String toString() {
             return "LoadSummary(" +
                 "startTimeMs=" + startTimeMs +
                 ", endTimeMs=" + endTimeMs +
+                ", schedulerQueueTimeMs=" + schedulerQueueTimeMs +
                 ", numRecords=" + numRecords +
                 ", numBytes=" + numBytes + ")";
         }

diff --git a/...rdinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java b/...rdinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java
@@ -1630,10 +1630,12 @@ public void scheduleLoadOperation(
                                             ctx.transitionTo(CoordinatorState.ACTIVE);
                                             if (summary != null) {
                                                 runtimeMetrics.recordPartitionLoadSensor(summary.startTimeMs(), summary.endTimeMs());
+                                                log.info("Finished loading of metadata from {} in {}ms with epoch {} where {}ms " +
+                                                    "was spent in the scheduler. Loaded {} records which total to {} bytes.",
+                                                    summary.endTimeMs() - summary.startTimeMs(), tp, partitionEpoch,
+                                                    summary.schedulerQueueTimeMs(), summary.numRecords(), summary.numBytes()
+                                                );
                                             }
-                                            log.info("Finished loading of metadata from {} with epoch {} and LoadSummary={}.",
-                                                tp, partitionEpoch, summary
-                                            );
                                         } catch (Throwable ex) {
                                             log.error("Failed to load metadata from {} with epoch {} due to {}.",
                                                 tp, partitionEpoch, ex.toString()

diff --git a/...ator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java b/...ator/src/test/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntimeTest.java
@@ -2331,6 +2331,7 @@ public void testPartitionLoadSensor() {
                     new CoordinatorLoader.LoadSummary(
                         startTimeMs,
                         startTimeMs + 1000,
+                        startTimeMs + 500,
                         30,
                         3000),
                     Collections.emptyList(),
@@ -2385,6 +2386,7 @@ public void testPartitionLoadGeneratesSnapshotAtHighWatermark() {
                     new CoordinatorLoader.LoadSummary(
                         1000,
                         2000,
+                        1500,
                         30,
                         3000),
                     Arrays.asList(5L, 15L, 27L),
@@ -2440,6 +2442,7 @@ public void testPartitionLoadGeneratesSnapshotAtHighWatermarkNoRecordsLoaded() {
                     new CoordinatorLoader.LoadSummary(
                         1000,
                         2000,
+                        1500,
                         30,
                         3000),
                     Collections.emptyList(),