(kafka) branch 4.1 updated: KAFKA-19522: avoid electing fenced lastKnownLeader (#20200)

[frankvicky]

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

frankvicky pushed a commit to branch 4.1
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/4.1 by this push:
     new e4e2dce2ebf KAFKA-19522: avoid electing fenced lastKnownLeader (#20200)
e4e2dce2ebf is described below

commit e4e2dce2ebfa7ac05fad4816936925342e6902ee
Author: Calvin Liu <83986057+calvinconflu...@users.noreply.github.com>
AuthorDate: Sun Jul 20 01:16:37 2025 -0700

    KAFKA-19522: avoid electing fenced lastKnownLeader (#20200)
    
    This patch fixes the bug that allows the last known leader to be elected as 
a partition leader while still in a fenced state, before the next heartbeat 
removes the fence.
    https://issues.apache.org/jira/browse/KAFKA-19522
    
    Reviewers: Jun Rao <jun...@gmail.com>, TengYao Chi
    <frankvi...@apache.org>
---
 .../kafka/controller/PartitionChangeBuilder.java   |  3 +-
 .../controller/PartitionChangeBuilderTest.java     | 36 ++++++++++++++++++++++
 2 files changed, 38 insertions(+), 1 deletion(-)

diff --git 
a/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java
 
b/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java
index ea196dfc741..a1d35293bd1 100644
--- 
a/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java
+++ 
b/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java
@@ -311,9 +311,10 @@ public class PartitionChangeBuilder {
                     topicId, partitionId, 
Arrays.toString(partition.lastKnownElr));
             return false;
         }
-        if (isAcceptableLeader.test(partition.lastKnownElr[0])) {
+        if (!isAcceptableLeader.test(partition.lastKnownElr[0])) {
             log.trace("Try to elect last known leader for {}-{} but last known 
leader is not alive. last known leader={}",
                     topicId, partitionId, partition.lastKnownElr[0]);
+            return false;
         }
         return true;
     }
diff --git 
a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
 
b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
index bd7de9bbd15..f836be9e93f 100644
--- 
a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
+++ 
b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java
@@ -34,6 +34,7 @@ import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.EnumSource;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 
@@ -1233,4 +1234,39 @@ public class PartitionChangeBuilderTest {
         // No change to the partition.
         assertEquals(Optional.empty(), builder.build());
     }
+
+    @ParameterizedTest
+    @EnumSource(value = Election.class)
+    public void testEligibleLeaderReplicas_NotEligibleLastKnownLeader(Election 
type) {
+        short version = 2;
+        PartitionRegistration partition = new PartitionRegistration.Builder()
+                .setReplicas(new int[] {1, 2, 3, 4})
+                .setDirectories(new Uuid[]{
+                        Uuid.fromString("zANDdMukTEqefOvHpmniMg"),
+                        Uuid.fromString("Ui2Eq8rbRiuW7m7uiPTRyg"),
+                        Uuid.fromString("MhgJOZrrTsKNcGM0XKK4aA"),
+                        Uuid.fromString("Y25PaCAmRfyGIKxAThhBAw")
+                })
+                .setIsr(new int[] {})
+                .setElr(new int[] {})
+                .setLastKnownElr(new int[] {1})
+                .setLeader(-1)
+                .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED)
+                .setLeaderEpoch(100)
+                .setPartitionEpoch(200)
+                .build();
+        Uuid topicId = Uuid.fromString("FbrrdcfiR-KC2CPSTHaJrg");
+
+        PartitionChangeBuilder builder = new PartitionChangeBuilder(partition, 
topicId, 0, r -> false,
+                metadataVersionForPartitionChangeRecordVersion(version), 3)
+            .setElection(type)
+            .setEligibleLeaderReplicasEnabled(true)
+            .setDefaultDirProvider(DEFAULT_DIR_PROVIDER)
+            .setUseLastKnownLeaderInBalancedRecovery(true);
+
+        builder.setTargetIsr(List.of());
+
+        // No change to the partition.
+        assertEquals(Optional.empty(), builder.build());
+    }
 }