KAFKA-15039: Reduce logging level to trace in PartitionChangeBuilder.… (#13780)

…tryElection()

A CPU profile in a large cluster showed PartitionChangeBuilder.tryElection() taking significant CPU due to logging. We adjust the logging statements in that method for clean elections from DEBUG level to TRACE to mitigate the impact of this logging under normal operations.  Unclean elections are now logged at the INFO level rather than DEBUG.

Reviewers: Jason Gustafson <jason@confluent.io>, Colin P. McCabe <cmccabe@apache.org>

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

@@ -220,13 +220,16 @@ public class PartitionChangeBuilder {
     private void tryElection(PartitionChangeRecord record) {
         ElectionResult electionResult = electLeader();
         if (electionResult.node != partition.leader) {
-            log.debug(
-                "Setting new leader for topicId {}, partition {} to {} using {} election",
-                topicId,
-                partitionId,
-                electionResult.node,
-                electionResult.unclean ? "an unclean" : "a clean"
-            );
+            // generating log messages for partition elections can get expensive on large clusters,
+            // so only log clean elections at TRACE level; log unclean elections at INFO level
+            // to ensure the message is emitted since an unclean election can lead to data loss.
+            if (electionResult.unclean) {
+                log.info("Setting new leader for topicId {}, partition {} to {} using an unclean election",
+                    topicId, partitionId, electionResult.node);
+            } else {
+                log.trace("Setting new leader for topicId {}, partition {} to {} using a clean election",
+                    topicId, partitionId, electionResult.node);
+            }
             record.setLeader(electionResult.node);
             if (electionResult.unclean) {
                 // If the election was unclean, we have to forcibly set the ISR to just the
@@ -239,7 +242,7 @@ public class PartitionChangeBuilder {
                 }
             }
         } else {
-            log.debug("Failed to find a new leader with current state: {}", this);
+            log.trace("Failed to find a new leader with current state: {}", this);
         }
     }