make sure beginQuorumEpochRequst send to non acks node

Author: TaiJuWu

raft/src/main/java/org/apache/kafka/raft/LeaderState.java

@@ -192,7 +192,8 @@ public Set<ReplicaKey> needToSendBeginQuorumRequests(long currentTimeMs) {
         Set<ReplicaKey> replicaKeys = new HashSet<>();
         beginQuorumEpochTimer.update(currentTimeMs);
         for (ReplicaState state : voterStates.values()) {
-            if (beginQuorumEpochTimer.currentTimeMs() - state.lastFetchTimestamp >= beginQuorumEpochTimeoutMs) {
+            if (beginQuorumEpochTimer.currentTimeMs() - state.lastFetchTimestamp >= beginQuorumEpochTimeoutMs
+                || !state.hasAcknowledgedLeader) {
                 replicaKeys.add(state.replicaKey());
             }
         }
@@ -963,6 +964,10 @@ private void updateVoterAndObserverStates(VoterSet lastVoterSet) {
         kafkaRaftMetrics.updateNumObservers(observerStates.size());
     }
 
+    private static class VoterState {
+
+    }
+
     public static class ReplicaState implements Comparable<ReplicaState> {
         private ReplicaKey replicaKey;
         private Endpoints listeners;