apache · hachikuji · Dec 9, 2020 · Oct 31, 2020
diff --git a/clients/src/main/resources/common/message/LeaderChangeMessage.json b/clients/src/main/resources/common/message/LeaderChangeMessage.json
@@ -22,7 +22,9 @@
     {"name": "LeaderId", "type": "int32", "versions": "0+",
       "about": "The ID of the newly elected leader"},
     {"name": "Voters", "type": "[]Voter", "versions": "0+",
-      "about": "The voters who voted for the current leader"}
+      "about": "The set of voters in the quorum for this epoch"},
+    {"name": "GrantingVoters", "type": "[]Voter", "versions": "0+",
+      "about": "The voters who voted for the leader at the time of election"}
   ],
   "commonStructs": [
     { "name": "Voter", "versions": "0+", "fields": [

diff --git a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
@@ -314,7 +314,7 @@ public void initialize() throws IOException {
 
         long currentTimeMs = time.milliseconds();
         if (quorum.isLeader()) {
-            onBecomeLeader(currentTimeMs);
+            throw new IllegalStateException("Voter cannot initialize as a Leader");
         } else if (quorum.isCandidate()) {
             onBecomeCandidate(currentTimeMs);
         } else if (quorum.isFollower()) {
@@ -371,14 +371,23 @@ private void onBecomeLeader(long currentTimeMs) {
         );
     }
 
-    private void appendLeaderChangeMessage(LeaderState state, long currentTimeMs) {
-        List<Voter> voters = state.followers().stream()
+    private static List<Voter> convertToVoters(Set<Integer> voterIds) {
+        return voterIds.stream()
             .map(follower -> new Voter().setVoterId(follower))
             .collect(Collectors.toList());
+    }
+
+    private void appendLeaderChangeMessage(LeaderState state, long currentTimeMs) {
+        List<Voter> voters = convertToVoters(state.followers());
+        List<Voter> grantingVoters = convertToVoters(state.grantingVoters());
+
+        // Adding the leader to the voters as any voter always votes for itself.
+        voters.add(new Voter().setVoterId(state.election().leaderId()));
 
         LeaderChangeMessage leaderChangeMessage = new LeaderChangeMessage()
             .setLeaderId(state.election().leaderId())
-            .setVoters(voters);
+            .setVoters(voters)
+            .setGrantingVoters(grantingVoters);
 
         MemoryRecords records = MemoryRecords.withLeaderChangeMessage(
             currentTimeMs, quorum.epoch(), leaderChangeMessage);
@@ -1556,7 +1565,7 @@ private long pollLeader(long currentTimeMs) {
 
         long timeUntilSend = maybeSendRequests(
             currentTimeMs,
-            state.nonEndorsingFollowers(),
+            state.nonEndorsingVoters(),
             this::buildBeginQuorumEpochRequest
         );
 

diff --git a/raft/src/main/java/org/apache/kafka/raft/LeaderState.java b/raft/src/main/java/org/apache/kafka/raft/LeaderState.java
@@ -34,13 +34,15 @@ public class LeaderState implements EpochState {
     private Optional<LogOffsetMetadata> highWatermark;
     private final Map<Integer, VoterState> voterReplicaStates = new HashMap<>();
     private final Map<Integer, ReplicaState> observerReplicaStates = new HashMap<>();
+    private final Set<Integer> grantingVoters = new HashSet<>();
     private static final long OBSERVER_SESSION_TIMEOUT_MS = 300_000L;
 
     protected LeaderState(
         int localId,
         int epoch,
         long epochStartOffset,
-        Set<Integer> voters
+        Set<Integer> voters,
+        Set<Integer> grantingVoters
     ) {
         this.localId = localId;
         this.epoch = epoch;
@@ -51,6 +53,7 @@ protected LeaderState(
             boolean hasEndorsedLeader = voterId == localId;
             this.voterReplicaStates.put(voterId, new VoterState(voterId, hasEndorsedLeader));
         }
+        this.grantingVoters.addAll(grantingVoters);
     }
 
     @Override
@@ -72,11 +75,15 @@ public Set<Integer> followers() {
         return voterReplicaStates.keySet().stream().filter(id -> id != localId).collect(Collectors.toSet());
     }
 
+    public Set<Integer> grantingVoters() {
+        return this.grantingVoters;
+    }
+
     public int localId() {
         return localId;
     }
 
-    public Set<Integer> nonEndorsingFollowers() {
+    public Set<Integer> nonEndorsingVoters() {
         Set<Integer> nonEndorsing = new HashSet<>();
         for (VoterState state : voterReplicaStates.values()) {
             if (!state.hasEndorsedLeader)

diff --git a/raft/src/main/java/org/apache/kafka/raft/QuorumState.java b/raft/src/main/java/org/apache/kafka/raft/QuorumState.java
@@ -420,7 +420,8 @@ public void transitionToLeader(long epochStartOffset) throws IOException {
             localId,
             epoch(),
             epochStartOffset,
-            voters
+            voters,
+            candidateState.grantingVoters()
         ));
     }
 

diff --git a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
@@ -165,7 +165,7 @@ public void testInitializeAsResignedLeaderFromStateStore() throws Exception {
         context.time.sleep(context.electionTimeoutMs);
         context.pollUntilSend();
         context.assertVotedCandidate(epoch + 1, localId);
-        context.assertSentVoteRequest(epoch + 1, 0, 0L);
+        context.assertSentVoteRequest(epoch + 1, 0, 0L, 1);
     }
 
     @Test
@@ -238,7 +238,7 @@ public void testInitializeAsCandidateAndBecomeLeader() throws Exception {
         context.pollUntilSend();
         context.assertVotedCandidate(1, context.localId);
 
-        int correlationId = context.assertSentVoteRequest(1, 0, 0L);
+        int correlationId = context.assertSentVoteRequest(1, 0, 0L, 1);
         context.deliverResponse(correlationId, otherNodeId, context.voteResponse(true, Optional.empty(), 1));
 
         // Become leader after receiving the vote
@@ -252,16 +252,56 @@ public void testInitializeAsCandidateAndBecomeLeader() throws Exception {
 
         // Send BeginQuorumEpoch to voters
         context.client.poll();
-        context.assertSentBeginQuorumEpochRequest(1);
+        context.assertSentBeginQuorumEpochRequest(1, 1);
 
         Records records = context.log.read(0, Isolation.UNCOMMITTED).records;
         RecordBatch batch = records.batches().iterator().next();
         assertTrue(batch.isControlBatch());
 
         Record record = batch.iterator().next();
         assertEquals(electionTimestamp, record.timestamp());
-        RaftClientTestContext.verifyLeaderChangeMessage(context.localId,
-            Collections.singletonList(otherNodeId), record.key(), record.value());
+        RaftClientTestContext.verifyLeaderChangeMessage(context.localId, Arrays.asList(otherNodeId, context.localId),
+            Arrays.asList(otherNodeId, context.localId), record.key(), record.value());
+    }
+
+    @Test
+    public void testInitializeAsCandidateAndBecomeLeaderQuorumOfThree() throws Exception {
+        int localId = 0;
+        final int firstNodeId = 1;
+        final int secondNodeId = 2;
+        Set<Integer> voters = Utils.mkSet(localId, firstNodeId, secondNodeId);
+        RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters).build();
+
+        context.assertUnknownLeader(0);
+        context.time.sleep(2 * context.electionTimeoutMs);
+
+        context.pollUntilSend();
+        context.assertVotedCandidate(1, context.localId);
+
+        int correlationId = context.assertSentVoteRequest(1, 0, 0L, 2);
+        context.deliverResponse(correlationId, firstNodeId, context.voteResponse(true, Optional.empty(), 1));
+
+        // Become leader after receiving the vote
+        context.client.poll();
+        context.assertElectedLeader(1, context.localId);
+        long electionTimestamp = context.time.milliseconds();
+
+        // Leader change record appended
+        assertEquals(1L, context.log.endOffset().offset);
+        assertEquals(1L, context.log.lastFlushedOffset());
+
+        // Send BeginQuorumEpoch to voters
+        context.client.poll();
+        context.assertSentBeginQuorumEpochRequest(1, 2);
+
+        Records records = context.log.read(0, Isolation.UNCOMMITTED).records;
+        RecordBatch batch = records.batches().iterator().next();
+        assertTrue(batch.isControlBatch());
+
+        Record record = batch.iterator().next();
+        assertEquals(electionTimestamp, record.timestamp());
+        RaftClientTestContext.verifyLeaderChangeMessage(context.localId, Arrays.asList(firstNodeId, secondNodeId, context.localId),
+            Arrays.asList(firstNodeId, context.localId), record.key(), record.value());
     }
 
     @Test
@@ -548,11 +588,11 @@ public void testVoteRequestTimeout() throws Exception {
         context.pollUntilSend();
         context.assertVotedCandidate(epoch, context.localId);
 
-        int correlationId = context.assertSentVoteRequest(epoch, 0, 0L);
+        int correlationId = context.assertSentVoteRequest(epoch, 0, 0L, 1);
 
         context.time.sleep(context.requestTimeoutMs);
         context.client.poll();
-        int retryCorrelationId = context.assertSentVoteRequest(epoch, 0, 0L);
+        int retryCorrelationId = context.assertSentVoteRequest(epoch, 0, 0L, 1);
 
         // Even though we have resent the request, we should still accept the response to
         // the first request if it arrives late.
@@ -760,7 +800,7 @@ public void testRetryElection() throws Exception {
         context.assertVotedCandidate(epoch, context.localId);
 
         // Quorum size is two. If the other member rejects, then we need to schedule a revote.
-        int correlationId = context.assertSentVoteRequest(epoch, 0, 0L);
+        int correlationId = context.assertSentVoteRequest(epoch, 0, 0L, 1);
         context.deliverResponse(correlationId, otherNodeId, context.voteResponse(false, Optional.empty(), 1));
 
         context.client.poll();
@@ -779,7 +819,7 @@ public void testRetryElection() throws Exception {
         context.client.poll();
         context.pollUntilSend();
         context.assertVotedCandidate(epoch + 1, context.localId);
-        context.assertSentVoteRequest(epoch + 1, 0, 0L);
+        context.assertSentVoteRequest(epoch + 1, 0, 0L, 1);
     }
 
     @Test
@@ -840,7 +880,7 @@ public void testVoterBecomeCandidateAfterFetchTimeout() throws Exception {
 
         context.pollUntilSend();
 
-        context.assertSentVoteRequest(epoch + 1, lastEpoch, 1L);
+        context.assertSentVoteRequest(epoch + 1, lastEpoch, 1L, 1);
         context.assertVotedCandidate(epoch + 1, context.localId);
     }
 
@@ -1565,7 +1605,7 @@ public void testFetchShouldBeTreatedAsLeaderEndorsement() throws Exception {
         context.pollUntilSend();
 
         // We send BeginEpoch, but it gets lost and the destination finds the leader through the Fetch API
-        context.assertSentBeginQuorumEpochRequest(epoch);
+        context.assertSentBeginQuorumEpochRequest(epoch, 1);
 
         context.deliverRequest(context.fetchRequest(
             epoch, otherNodeId, 0L, 0, 500));
@@ -1624,8 +1664,8 @@ public void testLeaderAppendSingleMemberQuorum() throws IOException {
 
         Record record = readRecords.get(0);
         assertEquals(now, record.timestamp());
-        RaftClientTestContext.verifyLeaderChangeMessage(context.localId, Collections.emptyList(),
-            record.key(), record.value());
+        RaftClientTestContext.verifyLeaderChangeMessage(context.localId, Collections.singletonList(context.localId),
+            Collections.singletonList(context.localId), record.key(), record.value());
 
         MutableRecordBatch batch = batches.get(1);
         assertEquals(1, batch.partitionLeaderEpoch());
@@ -1758,7 +1798,7 @@ public void testClusterAuthorizationFailedInBeginQuorumEpoch() throws Exception
         context.expectAndGrantVotes(epoch);
 
         context.pollUntilSend();
-        int correlationId = context.assertSentBeginQuorumEpochRequest(epoch);
+        int correlationId = context.assertSentBeginQuorumEpochRequest(epoch, 1);
         BeginQuorumEpochResponseData response = new BeginQuorumEpochResponseData()
             .setErrorCode(Errors.CLUSTER_AUTHORIZATION_FAILED.code());
 
@@ -1782,7 +1822,7 @@ public void testClusterAuthorizationFailedInVote() throws Exception {
         context.pollUntilSend();
         context.assertVotedCandidate(epoch, context.localId);
 
-        int correlationId = context.assertSentVoteRequest(epoch, 0, 0L);
+        int correlationId = context.assertSentVoteRequest(epoch, 0, 0L, 1);
         VoteResponseData response = new VoteResponseData()
             .setErrorCode(Errors.CLUSTER_AUTHORIZATION_FAILED.code());
 

diff --git a/raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java b/raft/src/test/java/org/apache/kafka/raft/LeaderStateTest.java
@@ -39,32 +39,32 @@ public class LeaderStateTest {
     public void testFollowerEndorsement() {
         int node1 = 1;
         int node2 = 2;
-        LeaderState state = new LeaderState(localId, epoch, 0L, mkSet(localId, node1, node2));
-        assertEquals(mkSet(node1, node2), state.nonEndorsingFollowers());
+        LeaderState state = new LeaderState(localId, epoch, 0L, mkSet(localId, node1, node2), Collections.emptySet());
+        assertEquals(mkSet(node1, node2), state.nonEndorsingVoters());
         state.addEndorsementFrom(node1);
-        assertEquals(Collections.singleton(node2), state.nonEndorsingFollowers());
+        assertEquals(Collections.singleton(node2), state.nonEndorsingVoters());
         state.addEndorsementFrom(node2);
-        assertEquals(Collections.emptySet(), state.nonEndorsingFollowers());
+        assertEquals(Collections.emptySet(), state.nonEndorsingVoters());
     }
 
     @Test
     public void testNonFollowerEndorsement() {
         int nonVoterId = 1;
-        LeaderState state = new LeaderState(localId, epoch, 0L, Collections.singleton(localId));
+        LeaderState state = new LeaderState(localId, epoch, 0L, Collections.singleton(localId), Collections.emptySet());
         assertThrows(IllegalArgumentException.class, () -> state.addEndorsementFrom(nonVoterId));
     }
 
     @Test
     public void testUpdateHighWatermarkQuorumSizeOne() {
-        LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId));
+        LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId), Collections.emptySet());
         assertEquals(Optional.empty(), state.highWatermark());
         assertTrue(state.updateLocalState(0, new LogOffsetMetadata(15L)));
         assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());
     }
 
     @Test
     public void testNonMonotonicEndOffsetUpdate() {
-        LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId));
+        LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId), Collections.emptySet());
         assertEquals(Optional.empty(), state.highWatermark());
         assertTrue(state.updateLocalState(0, new LogOffsetMetadata(15L)));
         assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());
@@ -74,7 +74,7 @@ public void testNonMonotonicEndOffsetUpdate() {
 
     @Test
     public void testIdempotentEndOffsetUpdate() {
-        LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId));
+        LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId), Collections.emptySet());
         assertEquals(Optional.empty(), state.highWatermark());
         assertTrue(state.updateLocalState(0, new LogOffsetMetadata(15L)));
         assertFalse(state.updateLocalState(0, new LogOffsetMetadata(15L)));
@@ -83,7 +83,7 @@ public void testIdempotentEndOffsetUpdate() {
 
     @Test
     public void testUpdateHighWatermarkMetadata() {
-        LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId));
+        LeaderState state = new LeaderState(localId, epoch, 15L, Collections.singleton(localId), Collections.emptySet());
         assertEquals(Optional.empty(), state.highWatermark());
 
         LogOffsetMetadata initialHw = new LogOffsetMetadata(15L, Optional.of(new MockOffsetMetadata("foo")));
@@ -98,11 +98,11 @@ public void testUpdateHighWatermarkMetadata() {
     @Test
     public void testUpdateHighWatermarkQuorumSizeTwo() {
         int otherNodeId = 1;
-        LeaderState state = new LeaderState(localId, epoch, 10L, mkSet(localId, otherNodeId));
+        LeaderState state = new LeaderState(localId, epoch, 10L, mkSet(localId, otherNodeId), Collections.emptySet());
         assertFalse(state.updateLocalState(0, new LogOffsetMetadata(15L)));
         assertEquals(Optional.empty(), state.highWatermark());
         assertTrue(state.updateReplicaState(otherNodeId, 0, new LogOffsetMetadata(10L)));
-        assertEquals(Collections.emptySet(), state.nonEndorsingFollowers());
+        assertEquals(Collections.emptySet(), state.nonEndorsingVoters());
         assertEquals(Optional.of(new LogOffsetMetadata(10L)), state.highWatermark());
         assertTrue(state.updateReplicaState(otherNodeId, 0, new LogOffsetMetadata(15L)));
         assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());
@@ -111,7 +111,7 @@ public void testUpdateHighWatermarkQuorumSizeTwo() {
     @Test
     public void testHighWatermarkUnknownUntilStartOfLeaderEpoch() {
         int otherNodeId = 1;
-        LeaderState state = new LeaderState(localId, epoch, 15L, mkSet(localId, otherNodeId));
+        LeaderState state = new LeaderState(localId, epoch, 15L, mkSet(localId, otherNodeId), Collections.emptySet());
         assertFalse(state.updateLocalState(0, new LogOffsetMetadata(20L)));
         assertEquals(Optional.empty(), state.highWatermark());
         assertFalse(state.updateReplicaState(otherNodeId, 0, new LogOffsetMetadata(10L)));
@@ -124,11 +124,11 @@ public void testHighWatermarkUnknownUntilStartOfLeaderEpoch() {
     public void testUpdateHighWatermarkQuorumSizeThree() {
         int node1 = 1;
         int node2 = 2;
-        LeaderState state = new LeaderState(localId, epoch, 10L, mkSet(localId, node1, node2));
+        LeaderState state = new LeaderState(localId, epoch, 10L, mkSet(localId, node1, node2), Collections.emptySet());
         assertFalse(state.updateLocalState(0, new LogOffsetMetadata(15L)));
         assertEquals(Optional.empty(), state.highWatermark());
         assertTrue(state.updateReplicaState(node1, 0, new LogOffsetMetadata(10L)));
-        assertEquals(Collections.singleton(node2), state.nonEndorsingFollowers());
+        assertEquals(Collections.singleton(node2), state.nonEndorsingVoters());
         assertEquals(Optional.of(new LogOffsetMetadata(10L)), state.highWatermark());
         assertTrue(state.updateReplicaState(node2, 0, new LogOffsetMetadata(15L)));
         assertEquals(Optional.of(new LogOffsetMetadata(15L)), state.highWatermark());
@@ -173,7 +173,7 @@ private LeaderState setUpLeaderAndFollowers(int follower1,
                                                 int follower2,
                                                 long leaderStartOffset,
                                                 long leaderEndOffset) {
-        LeaderState state = new LeaderState(localId, epoch, leaderStartOffset, mkSet(localId, follower1, follower2));
+        LeaderState state = new LeaderState(localId, epoch, leaderStartOffset, mkSet(localId, follower1, follower2), Collections.emptySet());
         state.updateLocalState(0, new LogOffsetMetadata(leaderEndOffset));
         assertEquals(Optional.empty(), state.highWatermark());
         state.updateReplicaState(follower1, 0, new LogOffsetMetadata(leaderStartOffset));
@@ -186,7 +186,7 @@ public void testGetObserverStatesWithObserver() {
         int observerId = 10;
         long endOffset = 10L;
 
-        LeaderState state = new LeaderState(localId, epoch, endOffset, mkSet(localId));
+        LeaderState state = new LeaderState(localId, epoch, endOffset, mkSet(localId), Collections.emptySet());
         long timestamp = 20L;
         assertFalse(state.updateReplicaState(observerId, timestamp, new LogOffsetMetadata(endOffset)));
 
@@ -198,7 +198,7 @@ public void testNoOpForNegativeRemoteNodeId() {
         int observerId = -1;
         long endOffset = 10L;
 
-        LeaderState state = new LeaderState(localId, epoch, endOffset, mkSet(localId));
+        LeaderState state = new LeaderState(localId, epoch, endOffset, mkSet(localId), Collections.emptySet());
         assertFalse(state.updateReplicaState(observerId, 0, new LogOffsetMetadata(endOffset)));
 
         assertEquals(Collections.emptyMap(), state.getObserverStates(10));