KAFKA-18379: Enforce resigned cannot transition to any other state in…

… same epoch JIRA: KAFKA-18379
apache · Feb 14, 2025 · f0c6fa4 · f0c6fa4
1 parent e7a2af8
commit f0c6fa4
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 4 deletions.
diff --git a/raft/src/main/java/org/apache/kafka/raft/QuorumState.java b/raft/src/main/java/org/apache/kafka/raft/QuorumState.java
@@ -382,7 +382,7 @@ public void transitionToResigned(List<ReplicaKey> preferredSuccessors) {
      */
     public void transitionToUnattached(int epoch, OptionalInt leaderId) {
         int currentEpoch = state.epoch();
-        if (epoch < currentEpoch || (epoch == currentEpoch && !isProspective())) {
+        if (epoch < currentEpoch || (epoch == currentEpoch && (!isProspective() || isResigned()))) {
             throw new IllegalStateException(
                 String.format(
                     "Cannot transition to Unattached with epoch %d from current state %s",
@@ -576,6 +576,14 @@ public void transitionToFollower(int epoch, int leaderId, Endpoints endpoints) {
                         state
                     )
                 );
+            } else if (isResigned()) {
+                throw new IllegalStateException(
+                    String.format(
+                        "Cannot transition to Follower within same epoch %s from state %s",
+                        epoch,
+                        state
+                    )
+                );
             }
         }
 

diff --git a/raft/src/test/java/org/apache/kafka/raft/QuorumStateTest.java b/raft/src/test/java/org/apache/kafka/raft/QuorumStateTest.java
@@ -2505,7 +2505,6 @@ public void testLeaderToAnyStateLowerEpoch(KRaftVersion kraftVersion) {
     /**
      * Test transitions from Resigned
      */
-    // KAFKA-18379 to fill in the rest of the cases
     @ParameterizedTest
     @EnumSource(value = KRaftVersion.class)
     public void testResignedToFollowerInSameEpoch(KRaftVersion kraftVersion) {
@@ -2519,8 +2518,7 @@ public void testResignedToFollowerInSameEpoch(KRaftVersion kraftVersion) {
         state.initialize(new OffsetAndEpoch(0L, logEndEpoch));
         assertTrue(state.isResigned());
         assertThrows(IllegalStateException.class, () -> state.transitionToFollower(epoch, localId, voters.listeners(localId)));
-        // KAFKA-18379 will fix this
-        state.transitionToFollower(epoch, node1, voters.listeners(node1));
+        assertThrows(IllegalStateException.class, () -> state.transitionToFollower(epoch, node1, voters.listeners(node1)));
     }
 
     @ParameterizedTest