(kafka) branch trunk updated: KAFKA-18940: fix electionWasClean (#19156)

[junrao]

This is an automated email from the ASF dual-hosted git repository.

junrao 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 db38bef0764 KAFKA-18940: fix electionWasClean (#19156)
db38bef0764 is described below

commit db38bef0764e999a3ab0d62fe7603bd2d61d4406
Author: Calvin Liu <83986057+calvinconflu...@users.noreply.github.com>
AuthorDate: Fri Mar 7 11:04:06 2025 -0800

    KAFKA-18940: fix electionWasClean (#19156)
    
    The electionWasClean should also consider if the election is done
    through ELR. Otherwise, the metric uncleanLeaderElection will wrongly
    count the ELR election
    https://issues.apache.org/jira/browse/KAFKA-18940
    
    Reviewers: Jun Rao <jun...@gmail.com>
---
 .../kafka/controller/metrics/ControllerMetricsChanges.java       | 3 ++-
 .../java/org/apache/kafka/metadata/PartitionRegistration.java    | 6 +++---
 .../org/apache/kafka/metadata/PartitionRegistrationTest.java     | 9 +++++----
 3 files changed, 10 insertions(+), 8 deletions(-)

diff --git 
a/metadata/src/main/java/org/apache/kafka/controller/metrics/ControllerMetricsChanges.java
 
b/metadata/src/main/java/org/apache/kafka/controller/metrics/ControllerMetricsChanges.java
index dc5b5cf5b93..7a4fef9182e 100644
--- 
a/metadata/src/main/java/org/apache/kafka/controller/metrics/ControllerMetricsChanges.java
+++ 
b/metadata/src/main/java/org/apache/kafka/controller/metrics/ControllerMetricsChanges.java
@@ -128,7 +128,8 @@ class ControllerMetricsChanges {
             isWithoutPreferredLeader = !next.hasPreferredLeader();
             // take current all replicas as ISR if prev is null (new created 
partition), so we won't treat it as unclean election.
             int[] prevIsr = prev != null ? prev.isr : next.replicas;
-            if (!PartitionRegistration.electionWasClean(next.leader, prevIsr)) 
{
+            int[] prevElr = prev != null ? prev.elr : new int[]{};
+            if (!PartitionRegistration.electionWasClean(next.leader, prevIsr, 
prevElr)) {
                 uncleanLeaderElection++;
             }
         }
diff --git 
a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java 
b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java
index 76f4fafe605..3891b624226 100644
--- 
a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java
+++ 
b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java
@@ -165,8 +165,8 @@ public class PartitionRegistration {
     public final int leaderEpoch;
     public final int partitionEpoch;
 
-    public static boolean electionWasClean(int newLeader, int[] isr) {
-        return newLeader == NO_LEADER || Replicas.contains(isr, newLeader);
+    public static boolean electionWasClean(int newLeader, int[] isr, int[] 
elr) {
+        return newLeader == NO_LEADER || Replicas.contains(isr, newLeader) || 
Replicas.contains(elr, newLeader);
     }
 
     private static List<Uuid> checkDirectories(PartitionRecord record) {
@@ -347,7 +347,7 @@ public class PartitionRegistration {
     }
 
     public void maybeLogPartitionChange(Logger log, String description, 
PartitionRegistration prev) {
-        if (!electionWasClean(leader, prev.isr)) {
+        if (!electionWasClean(leader, prev.isr, prev.elr)) {
             log.info("UNCLEAN partition change for {}: {}", description, 
diff(prev));
         } else if (log.isDebugEnabled()) {
             log.debug("partition change for {}: {}", description, diff(prev));
diff --git 
a/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java
 
b/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java
index 8dc817f2621..7d15ea32ddb 100644
--- 
a/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java
+++ 
b/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java
@@ -58,10 +58,11 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
 public class PartitionRegistrationTest {
     @Test
     public void testElectionWasClean() {
-        assertTrue(PartitionRegistration.electionWasClean(1, new int[]{1, 2}));
-        assertFalse(PartitionRegistration.electionWasClean(1, new int[]{0, 
2}));
-        assertFalse(PartitionRegistration.electionWasClean(1, new int[]{}));
-        assertTrue(PartitionRegistration.electionWasClean(3, new int[]{1, 2, 
3, 4, 5, 6}));
+        assertTrue(PartitionRegistration.electionWasClean(1, new int[]{1, 2}, 
new int[]{}));
+        assertFalse(PartitionRegistration.electionWasClean(1, new int[]{0, 2}, 
new int[]{}));
+        assertFalse(PartitionRegistration.electionWasClean(1, new int[]{}, new 
int[]{3, 4}));
+        assertTrue(PartitionRegistration.electionWasClean(3, new int[]{1, 2, 
3, 4, 5, 6}, new int[]{}));
+        assertTrue(PartitionRegistration.electionWasClean(3, new int[]{}, new 
int[]{1, 2, 3}));
     }
 
     @Test