KAFKA-16446: Improve controller event duration logging

core/src/main/scala/kafka/server/ControllerServer.scala

@@ -242,7 +242,8 @@ class ControllerServer(
           setDelegationTokenExpiryTimeMs(config.delegationTokenExpiryTimeMs).
           setDelegationTokenExpiryCheckIntervalMs(config.delegationTokenExpiryCheckIntervalMs).
           setUncleanLeaderElectionCheckIntervalMs(config.uncleanLeaderElectionCheckIntervalMs).
-          setInterBrokerListenerName(config.interBrokerListenerName.value())
+          setInterBrokerListenerName(config.interBrokerListenerName.value()).
+          setMinSlowEventTimeMs(config.minSlowEventTimeMs)
       }
       controller = controllerBuilder.build()
 

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -333,6 +333,7 @@ class KafkaConfig private(doLog: Boolean, val props: util.Map[_, _])
   val initialRegistrationTimeoutMs: Int = getInt(KRaftConfigs.INITIAL_BROKER_REGISTRATION_TIMEOUT_MS_CONFIG)
   val brokerHeartbeatIntervalMs: Int = getInt(KRaftConfigs.BROKER_HEARTBEAT_INTERVAL_MS_CONFIG)
   val brokerSessionTimeoutMs: Int = getInt(KRaftConfigs.BROKER_SESSION_TIMEOUT_MS_CONFIG)
+  val minSlowEventTimeMs: Int = getInt(KRaftConfigs.MIN_SLOW_EVENT_TIME_MS_CONFIG)
 
   def requiresZookeeper: Boolean = processRoles.isEmpty
   def usesSelfManagedQuorum: Boolean = processRoles.nonEmpty

metadata/src/main/java/org/apache/kafka/controller/QuorumController.java

@@ -150,6 +150,7 @@
 import static java.util.concurrent.TimeUnit.MICROSECONDS;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
+import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.kafka.controller.QuorumController.ControllerOperationFlag.DOES_NOT_UPDATE_QUEUE_TIME;
 
 
@@ -175,16 +176,21 @@
  */
 public final class QuorumController implements Controller {
     /**
-     * The maximum records that the controller will write in a single batch.
+     * The default maximum records that the controller will write in a single batch.
      */
-    private static final int MAX_RECORDS_PER_BATCH = 10000;
+    private static final int DEFAULT_MAX_RECORDS_PER_BATCH = 10000;
+
+    /**
+     * The default minimum event time that can be logged as a slow event.
+     */
+    private static final int DEFAULT_MIN_SLOW_EVENT_TIME_MS = 200;
 
     /**
      * The maximum records any user-initiated operation is allowed to generate.
      *
      * For now, this is set to the maximum records in a single batch.
      */
-    static final int MAX_RECORDS_PER_USER_OP = MAX_RECORDS_PER_BATCH;
+    static final int MAX_RECORDS_PER_USER_OP = DEFAULT_MAX_RECORDS_PER_BATCH;
 
     /**
      * A builder class which creates the QuorumController.
@@ -213,7 +219,8 @@ public static class Builder {
         private ConfigurationValidator configurationValidator = ConfigurationValidator.NO_OP;
         private Map<String, Object> staticConfig = Collections.emptyMap();
         private BootstrapMetadata bootstrapMetadata = null;
-        private int maxRecordsPerBatch = MAX_RECORDS_PER_BATCH;
+        private int maxRecordsPerBatch = DEFAULT_MAX_RECORDS_PER_BATCH;
+        private int minSlowEventTimeMs = DEFAULT_MIN_SLOW_EVENT_TIME_MS;
         private DelegationTokenCache tokenCache;
         private String tokenSecretKeyString;
         private long delegationTokenMaxLifeMs;
@@ -321,6 +328,11 @@ public Builder setMaxRecordsPerBatch(int maxRecordsPerBatch) {
             return this;
         }
 
+        public Builder setMinSlowEventTimeMs(int minSlowEventTimeMs) {
+            this.minSlowEventTimeMs = minSlowEventTimeMs;
+            return this;
+        }
+
         public Builder setCreateTopicPolicy(Optional<CreateTopicPolicy> createTopicPolicy) {
             this.createTopicPolicy = createTopicPolicy;
             return this;
@@ -433,7 +445,8 @@ public QuorumController build() throws Exception {
                     delegationTokenExpiryTimeMs,
                     delegationTokenExpiryCheckIntervalMs,
                     uncleanLeaderElectionCheckIntervalMs,
-                    interBrokerListenerName
+                    interBrokerListenerName,
+                    minSlowEventTimeMs
                 );
             } catch (Exception e) {
                 Utils.closeQuietly(queue, "event queue");
@@ -524,6 +537,7 @@ private void handleEventEnd(String name, long startProcessingTimeNs) {
         long deltaNs = endProcessingTime - startProcessingTimeNs;
         log.debug("Processed {} in {} us", name,
             MICROSECONDS.convert(deltaNs, NANOSECONDS));
+        slowEventsLogger.maybeLogEvent(name, deltaNs);
         controllerMetrics.updateEventQueueProcessingTime(NANOSECONDS.toMillis(deltaNs));
     }
 
@@ -1446,6 +1460,8 @@ private void replay(ApiMessage message, Optional<OffsetAndEpoch> snapshotId, lon
      */
     private final RecordRedactor recordRedactor;
 
+    private final SlowEventsLogger slowEventsLogger;
+
     private QuorumController(
         FaultHandler nonFatalFaultHandler,
         FaultHandler fatalFaultHandler,
@@ -1477,7 +1493,8 @@ private QuorumController(
         long delegationTokenExpiryTimeMs,
         long delegationTokenExpiryCheckIntervalMs,
         long uncleanLeaderElectionCheckIntervalMs,
-        String interBrokerListenerName
+        String interBrokerListenerName,
+        int minSlowEventTimeMs
     ) {
         this.nonFatalFaultHandler = nonFatalFaultHandler;
         this.fatalFaultHandler = fatalFaultHandler;
@@ -1587,7 +1604,7 @@ private QuorumController(
         }
         registerElectUnclean(TimeUnit.MILLISECONDS.toNanos(uncleanLeaderElectionCheckIntervalMs));
         registerExpireDelegationTokens(MILLISECONDS.toNanos(delegationTokenExpiryCheckIntervalMs));
-
+        registerUpdateSlowEventLogger(SECONDS.toNanos(30));
         // OffsetControlManager must be initialized last, because its constructor will take the
         // initial in-memory snapshot of all extant timeline data structures.
         this.offsetControl = new OffsetControlManager.Builder().
@@ -1599,6 +1616,8 @@ private QuorumController(
         log.info("Creating new QuorumController with clusterId {}", clusterId);
 
         this.raftClient.register(metaLogListener);
+        this.slowEventsLogger = new SlowEventsLogger(minSlowEventTimeMs,
+            controllerMetrics::getEventQueueProcessingTime99, logContext);
     }
 
     /**
@@ -1622,6 +1641,16 @@ private void registerWriteNoOpRecord(long maxIdleIntervalNs) {
             EnumSet.noneOf(PeriodicTaskFlag.class)));
     }
 
+    private void registerUpdateSlowEventLogger(long maxSlowEventWindowNs) {
+        periodicControl.registerTask(new PeriodicTask("updateSlowEventLoggerP99",
+            () -> {
+                slowEventsLogger.refreshPercentile();
+                return ControllerResult.of(Collections.emptyList(), false);
+            },
+            maxSlowEventWindowNs,
+            EnumSet.noneOf(PeriodicTaskFlag.class)));
+    }
+
     /**
      * Calculate what the period should be for the maybeFenceStaleBroker task.
      *

metadata/src/main/java/org/apache/kafka/controller/SlowEventsLogger.java

@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.controller;
+
+import org.apache.kafka.common.utils.LogContext;
+
+import org.slf4j.Logger;
+
+import java.util.function.Supplier;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static java.util.concurrent.TimeUnit.NANOSECONDS;
+
+/**
+ * Track the p99 for controller event queue processing time. If we encounter an event that takes longer
+ * than this cached p99 time, we will log it at INFO level on the controller logger.
+ */
+public class SlowEventsLogger {
+    /**
+     * Don't report any p99 events below this threshold. This prevents the controller from reporting p99 event
+     * times in the idle case where p99 event times are essentially the average as well.
+     */
+    private final long minSlowEventTimeNs;
+
+    /**
+     * Function that returns the current p99 time in millis. This call can be expensive, and since the histogram is
+     * biased towards the last 5 minutes of data, we only need to update this p99 every so often.
+     */
+    private final Supplier<Double> thresholdMsSupplier;
+
+    private final Logger log;
+
+    /**
+     * The current p99 threshold in nanos.
+     */
+    private long thresholdNs;
+
+    public SlowEventsLogger(
+        int minSlowEventTimeMs,
+        Supplier<Double> thresholdMsSupplier,
+        LogContext logContext
+    ) {
+        this.minSlowEventTimeNs = MILLISECONDS.toNanos(minSlowEventTimeMs);
+        this.thresholdMsSupplier = thresholdMsSupplier;
+        this.thresholdNs = minSlowEventTimeMs;
+        this.log = logContext.logger(SlowEventsLogger.class);
+    }
+
+    /**
+     * Produce an INFO log if the given event ran for at least as long as the current p99 event processing time.
+     *
+     * @return true if a slow event was logged, false otherwise.
+     */
+    public boolean maybeLogEvent(String name, long durationNs) {
+        if (durationNs >= minSlowEventTimeNs && durationNs >= thresholdNs) {
+            log.info("Slow controller event {} processed in {} us which is larger or equal to the p99 of {} us",
+                name,
+                NANOSECONDS.toMicros(durationNs),
+                NANOSECONDS.toMicros(thresholdNs)
+            );
+            return true;
+        }
+        return false;
+    }
+
+    public void refreshPercentile() {
+        thresholdNs = (long) (thresholdMsSupplier.get() * 1000000);
+        log.trace("Update slow controller event threshold (p99) to {} us.", NANOSECONDS.toMicros(thresholdNs));
+    }
+}

metadata/src/main/java/org/apache/kafka/controller/metrics/QuorumControllerMetrics.java

@@ -164,6 +164,15 @@ public void updateEventQueueProcessingTime(long durationMs) {
         eventQueueProcessingTimeUpdater.accept(durationMs);
     }
 
+    public double getEventQueueProcessingTime99() {
+        if (registry.isPresent()) {
+            Histogram histogram = registry.get().newHistogram(EVENT_QUEUE_PROCESSING_TIME_MS, true);
+            return histogram.getSnapshot().get99thPercentile();
+        } else {
+            // Only returned in unit tests when a metrics registry is not set.
+            return 0.0;
+        }
+    }
     public void setLastAppliedRecordOffset(long offset) {
         lastAppliedRecordOffset.set(offset);
     }

metadata/src/test/java/org/apache/kafka/controller/SlowEventsLoggerTest.java

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.controller;
+
+import org.apache.kafka.common.utils.LogContext;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+
+public class SlowEventsLoggerTest {
+    @Test
+    public void testSlowEvents() {
+        LogContext logContext = new LogContext();
+
+        AtomicReference<Double> p99 = new AtomicReference<>(0.0);
+        SlowEventsLogger logger = new SlowEventsLogger(100, p99::get, logContext);
+
+        // Initially, the p99 is zero
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(10)));
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(99)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(100)));
+
+
+        // Idle controller, low p99
+        p99.set(30.0);
+        logger.refreshPercentile();
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(90)));
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(99)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(100)));
+
+        // Busy controller, high p99
+        p99.set(1000.0);
+        logger.refreshPercentile();
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(100)));
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(200)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(1000)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(2000)));
+    }
+
+    @Test
+    public void testThresholdDisabled() {
+        LogContext logContext = new LogContext();
+
+        AtomicReference<Double> p99 = new AtomicReference<>(0.0);
+        // Set min slow event time to zero, effectively disabling the threshold
+        SlowEventsLogger logger = new SlowEventsLogger(0, p99::get, logContext);
+
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(0)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(10)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(99)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(100)));
+
+        p99.set(30.0);
+        logger.refreshPercentile();
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(0)));
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(29)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(30)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(100)));
+
+
+        p99.set(1000.0);
+        logger.refreshPercentile();
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(100)));
+        assertFalse(logger.maybeLogEvent("test", MILLISECONDS.toNanos(999)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(1000)));
+        assertTrue(logger.maybeLogEvent("test", MILLISECONDS.toNanos(2000)));
+    }
+}

server/src/main/java/org/apache/kafka/server/config/KRaftConfigs.java

@@ -114,6 +114,10 @@ public class KRaftConfigs {
     public static final String SERVER_MAX_STARTUP_TIME_MS_DOC = "The maximum number of milliseconds we will wait for the server to come up. " +
             "By default there is no limit. This should be used for testing only.";
 
+    public static final String MIN_SLOW_EVENT_TIME_MS_CONFIG = "controller.slow.event.min.ms";
+    public static final int MIN_SLOW_EVENT_TIME_MS_DEFAULT = 200;
+    public static final String MIN_SLOW_EVENT_TIME_MS_DOC = "Log controller events with a p99 duration slower than this amount.";
+
     public static final ConfigDef CONFIG_DEF =  new ConfigDef()
             .define(METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_CONFIG, LONG, METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES, atLeast(1), HIGH, METADATA_SNAPSHOT_MAX_NEW_RECORD_BYTES_DOC)
             .define(METADATA_SNAPSHOT_MAX_INTERVAL_MS_CONFIG, LONG, METADATA_SNAPSHOT_MAX_INTERVAL_MS_DEFAULT, atLeast(0), HIGH, METADATA_SNAPSHOT_MAX_INTERVAL_MS_DOC)
@@ -131,5 +135,6 @@ public class KRaftConfigs {
             .define(METADATA_MAX_RETENTION_BYTES_CONFIG, LONG, METADATA_MAX_RETENTION_BYTES_DEFAULT, null, HIGH, METADATA_MAX_RETENTION_BYTES_DOC)
             .define(METADATA_MAX_RETENTION_MILLIS_CONFIG, LONG, LogConfig.DEFAULT_RETENTION_MS, null, HIGH, METADATA_MAX_RETENTION_MILLIS_DOC)
             .define(METADATA_MAX_IDLE_INTERVAL_MS_CONFIG, INT, METADATA_MAX_IDLE_INTERVAL_MS_DEFAULT, atLeast(0), LOW, METADATA_MAX_IDLE_INTERVAL_MS_DOC)
+            .defineInternal(MIN_SLOW_EVENT_TIME_MS_CONFIG, INT, MIN_SLOW_EVENT_TIME_MS_DEFAULT, atLeast(0), MEDIUM, MIN_SLOW_EVENT_TIME_MS_DOC)
             .defineInternal(SERVER_MAX_STARTUP_TIME_MS_CONFIG, LONG, SERVER_MAX_STARTUP_TIME_MS_DEFAULT, atLeast(0), MEDIUM, SERVER_MAX_STARTUP_TIME_MS_DOC);
 }