diff --git a/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java b/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java index 25a1cf5ba8be7..53e152a9d3bab 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/migration/KRaftMigrationDriver.java @@ -518,21 +518,27 @@ public void run() throws Exception { Map dualWriteCounts = new TreeMap<>(); long startTime = time.nanoseconds(); + final long zkWriteTimeMs; if (isSnapshot) { zkMetadataWriter.handleSnapshot(image, countingOperationConsumer( dualWriteCounts, KRaftMigrationDriver.this::applyMigrationOperation)); - controllerMetrics.updateZkWriteSnapshotTimeMs(NANOSECONDS.toMillis(time.nanoseconds() - startTime)); + zkWriteTimeMs = NANOSECONDS.toMillis(time.nanoseconds() - startTime); + controllerMetrics.updateZkWriteSnapshotTimeMs(zkWriteTimeMs); } else { if (zkMetadataWriter.handleDelta(prevImage, image, delta, countingOperationConsumer( dualWriteCounts, KRaftMigrationDriver.this::applyMigrationOperation))) { // Only record delta write time if we changed something. Otherwise, no-op records will skew timings. - controllerMetrics.updateZkWriteDeltaTimeMs(NANOSECONDS.toMillis(time.nanoseconds() - startTime)); + zkWriteTimeMs = NANOSECONDS.toMillis(time.nanoseconds() - startTime); + controllerMetrics.updateZkWriteDeltaTimeMs(zkWriteTimeMs); + } else { + zkWriteTimeMs = 0; } } if (dualWriteCounts.isEmpty()) { log.trace("Did not make any ZK writes when handling KRaft {}", isSnapshot ? "snapshot" : "delta"); } else { - log.debug("Made the following ZK writes when handling KRaft {}: {}", isSnapshot ? "snapshot" : "delta", dualWriteCounts); + log.debug("Made the following ZK writes in {} ms when handling KRaft {}: {}", + zkWriteTimeMs, isSnapshot ? "snapshot" : "delta", dualWriteCounts); } // Persist the offset of the metadata that was written to ZK