(WIP) KAFKA-17243:MetadataQuorumCommand describe to include CommittedVoters

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -4684,6 +4684,10 @@ private QuorumInfo createQuorumResult(final DescribeQuorumResponseData.Partition
                     .map(this::translateReplicaState)
                     .collect(Collectors.toList());
 
+                List<QuorumInfo.ReplicaState> committedVoters = partition.committedVoters().stream()
+                    .map(this::translateReplicaState)
+                    .collect(Collectors.toList());
+
                 List<QuorumInfo.ReplicaState> observers = partition.observers().stream()
                     .map(this::translateReplicaState)
                     .collect(Collectors.toList());
@@ -4701,6 +4705,7 @@ private QuorumInfo createQuorumResult(final DescribeQuorumResponseData.Partition
                     partition.leaderEpoch(),
                     partition.highWatermark(),
                     voters,
+                    committedVoters,
                     observers,
                     nodes
                 );

clients/src/main/java/org/apache/kafka/clients/admin/QuorumInfo.java

@@ -31,6 +31,7 @@ public class QuorumInfo {
     private final long leaderEpoch;
     private final long highWatermark;
     private final List<ReplicaState> voters;
+    private final List<ReplicaState> committedVoters;
     private final List<ReplicaState> observers;
     private final Map<Integer, Node> nodes;
 
@@ -39,13 +40,15 @@ public class QuorumInfo {
         long leaderEpoch,
         long highWatermark,
         List<ReplicaState> voters,
+        List<ReplicaState> committedVoters,
         List<ReplicaState> observers,
         Map<Integer, Node> nodes
     ) {
         this.leaderId = leaderId;
         this.leaderEpoch = leaderEpoch;
         this.highWatermark = highWatermark;
         this.voters = voters;
+        this.committedVoters = committedVoters;
         this.observers = observers;
         this.nodes = nodes;
     }
@@ -66,6 +69,10 @@ public List<ReplicaState> voters() {
         return voters;
     }
 
+    public List<ReplicaState> committedVoters() {
+        return committedVoters;
+    }
+
     public List<ReplicaState> observers() {
         return observers;
     }

clients/src/main/resources/common/message/DescribeQuorumRequest.json

@@ -20,7 +20,8 @@
   "name": "DescribeQuorumRequest",
   // Version 1 adds additional fields in the response. The request is unchanged (KIP-836).
   // Version 2 adds additional fields in the response. The request is unchanged (KIP-853).
-  "validVersions": "0-2",
+  // Version 3 adds additional fields in the response. The request is unchanged (KIP-853).
+  "validVersions": "0-3",
   "flexibleVersions": "0+",
   "latestVersionUnstable": false,
   "fields": [

clients/src/main/resources/common/message/DescribeQuorumResponse.json

@@ -18,8 +18,9 @@
   "type": "response",
   "name": "DescribeQuorumResponse",
   // Version 1 adds LastFetchTimeStamp and LastCaughtUpTimestamp in ReplicaState (KIP-836).
-  // Version 2 adds ErrorMessage, Nodes, ErrorMessage in PartitionData, ReplicaDirectoryId in ReplicaState (KIP-853).
-  "validVersions": "0-2",
+  // Version 2 adds ErrorMessage, Nodes, ErrorMessage in ParitionData, ReplicaDirectoryId in ReplicaState (KIP-853).
+  // Version 3 adds CommittedVoters in PartitionData
+  "validVersions": "0-3",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "ErrorCode", "type": "int16", "versions": "0+",
@@ -46,6 +47,8 @@
           "about": "The high water mark."},
         { "name": "CurrentVoters", "type": "[]ReplicaState", "versions": "0+",
           "about": "The current voters of the partition."},
+        { "name": "CommittedVoters", "type": "[]ReplicaState", "versions": "3+", "ignorable": true,
+          "about":  "The voters has been committed."},
         { "name": "Observers", "type": "[]ReplicaState", "versions": "0+",
           "about": "The observers of the partition."}
       ]}

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -805,7 +805,12 @@ private static ApiVersionsResponse prepareApiVersionsResponseForDescribeFeatures
     }
 
     private static QuorumInfo defaultQuorumInfo(boolean emptyOptionals) {
-        return new QuorumInfo(1, 1, 1L,
+        return new QuorumInfo(1, 1L, 1L,
+                singletonList(new QuorumInfo.ReplicaState(1,
+                        emptyOptionals ? Uuid.ZERO_UUID : REPLICA_DIRECTORY_ID,
+                        100,
+                        emptyOptionals ? OptionalLong.empty() : OptionalLong.of(1000),
+                        emptyOptionals ? OptionalLong.empty() : OptionalLong.of(1000))),
                 singletonList(new QuorumInfo.ReplicaState(1,
                         emptyOptionals ? Uuid.ZERO_UUID : REPLICA_DIRECTORY_ID,
                         100,

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

@@ -1893,6 +1893,7 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
             leaderState.epoch(),
             leaderState.highWatermark().map(LogOffsetMetadata::offset).orElse(-1L),
             leaderState.voterStates().values(),
+            leaderState.committedVoterStates().values(),
             leaderState.observerStates(currentTimeMs).values(),
             currentTimeMs
         );

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

@@ -32,6 +32,7 @@
 import org.apache.kafka.raft.errors.NotLeaderException;
 import org.apache.kafka.raft.internals.AddVoterHandlerState;
 import org.apache.kafka.raft.internals.BatchAccumulator;
+import org.apache.kafka.raft.internals.KRaftControlRecordStateMachine;
 import org.apache.kafka.raft.internals.KRaftVersionUpgrade;
 import org.apache.kafka.raft.internals.KafkaRaftMetrics;
 import org.apache.kafka.raft.internals.RemoveVoterHandlerState;
@@ -72,9 +73,11 @@ public class LeaderState<T> implements EpochState {
     // This field is non-empty if the voter set at epoch start came from a snapshot or log segment
     private final OptionalLong offsetOfVotersAtEpochStart;
     private final KRaftVersion kraftVersionAtEpochStart;
+    private final KRaftControlRecordStateMachine partitionState;
 
     private Optional<LogOffsetMetadata> highWatermark = Optional.empty();
     private Map<Integer, ReplicaState> voterStates = new HashMap<>();
+    private Map<Integer, ReplicaState> committedVoterStates = new HashMap<>();
     private Optional<AddVoterHandlerState> addVoterHandlerState = Optional.empty();
     private Optional<RemoveVoterHandlerState> removeVoterHandlerState = Optional.empty();
 
@@ -120,7 +123,8 @@ protected LeaderState(
         BatchAccumulator<T> accumulator,
         int fetchTimeoutMs,
         LogContext logContext,
-        KafkaRaftMetrics kafkaRaftMetrics
+        KafkaRaftMetrics kafkaRaftMetrics,
+        KRaftControlRecordStateMachine partitionState
     ) {
         if (localVoterNode.voterKey().directoryId().isEmpty()) {
             throw new IllegalArgumentException(
@@ -139,14 +143,6 @@ protected LeaderState(
         this.localVoterNode = localVoterNode;
         this.epoch = epoch;
         this.epochStartOffset = epochStartOffset;
-
-        for (VoterSet.VoterNode voterNode: voterSetAtEpochStart.voterNodes()) {
-            boolean hasAcknowledgedLeader = voterNode.isVoter(localVoterNode.voterKey());
-            this.voterStates.put(
-                voterNode.voterKey().id(),
-                new ReplicaState(voterNode.voterKey(), hasAcknowledgedLeader, voterNode.listeners())
-            );
-        }
         this.grantingVoters = Set.copyOf(grantingVoters);
         this.log = logContext.logger(LeaderState.class);
         this.accumulator = Objects.requireNonNull(accumulator, "accumulator must be non-null");
@@ -158,6 +154,18 @@ protected LeaderState(
         this.voterSetAtEpochStart =  voterSetAtEpochStart;
         this.offsetOfVotersAtEpochStart = offsetOfVotersAtEpochStart;
         this.kraftVersionAtEpochStart = kraftVersionAtEpochStart;
+        this.partitionState = partitionState;
+
+        // When offset is empty or -1, it means the cluster is starting up.
+        boolean isInitializing = offsetOfVotersAtEpochStart.isEmpty() || offsetOfVotersAtEpochStart.getAsLong() == -1;
+        for (VoterSet.VoterNode voterNode: voterSetAtEpochStart.voterNodes()) {
+            boolean hasAcknowledgedLeader = voterNode.isVoter(localVoterNode.voterKey());
+            ReplicaState state = new ReplicaState(voterNode.voterKey(), hasAcknowledgedLeader, voterNode.listeners());
+            this.voterStates.put(voterNode.voterKey().id(), state);
+            if (isInitializing) {
+                this.committedVoterStates.put(voterNode.voterKey().id(), state);
+            }
+        }
 
         kafkaRaftMetrics.addLeaderMetrics();
         this.kafkaRaftMetrics = kafkaRaftMetrics;
@@ -704,6 +712,14 @@ Map<Integer, ReplicaState> voterStates() {
         return voterStates;
     }
 
+    Map<Integer, ReplicaState> committedVoterStates() {
+        if (highWatermark.isEmpty()) {
+            return Map.of();
+        }
+
+        return committedVoterStates;
+    }
+
     Map<ReplicaKey, ReplicaState> observerStates(final long currentTimeMs) {
         clearInactiveObservers(currentTimeMs);
         return observerStates;
@@ -752,6 +768,7 @@ private boolean maybeUpdateHighWatermark() {
                             !highWatermarkUpdateMetadata.metadata().equals(currentHighWatermarkMetadata.metadata()))) {
                         Optional<LogOffsetMetadata> oldHighWatermark = highWatermark;
                         highWatermark = highWatermarkUpdateOpt;
+                        updateCommittedVoter(currentHighWatermarkMetadata);
                         logHighWatermarkUpdate(
                             oldHighWatermark,
                             highWatermarkUpdateMetadata,
@@ -772,6 +789,7 @@ private boolean maybeUpdateHighWatermark() {
                 } else {
                     Optional<LogOffsetMetadata> oldHighWatermark = highWatermark;
                     highWatermark = highWatermarkUpdateOpt;
+                    updateCommittedVoter(highWatermarkUpdateMetadata);
                     logHighWatermarkUpdate(
                         oldHighWatermark,
                         highWatermarkUpdateMetadata,
@@ -821,7 +839,7 @@ public boolean updateLocalState(
         LogOffsetMetadata endOffsetMetadata,
         VoterSet lastVoterSet
     ) {
-        ReplicaState state = getOrCreateReplicaState(localVoterNode.voterKey());
+        ReplicaState state = getOrCreateObserverReplicaState(localVoterNode.voterKey());
         state.endOffset.ifPresent(currentEndOffset -> {
             if (currentEndOffset.offset() > endOffsetMetadata.offset()) {
                 throw new IllegalStateException("Detected non-monotonic update of local " +
@@ -858,7 +876,7 @@ public boolean updateReplicaState(
             );
         }
 
-        ReplicaState state = getOrCreateReplicaState(replicaKey);
+        ReplicaState state = getOrCreateObserverReplicaState(replicaKey);
 
         state.endOffset.ifPresent(currentEndOffset -> {
             if (currentEndOffset.offset() > fetchOffsetMetadata.offset()) {
@@ -867,7 +885,7 @@ public boolean updateReplicaState(
             }
         });
 
-        Optional<LogOffsetMetadata> leaderEndOffsetOpt = getOrCreateReplicaState(localVoterNode.voterKey()).endOffset;
+        Optional<LogOffsetMetadata> leaderEndOffsetOpt = getOrCreateObserverReplicaState(localVoterNode.voterKey()).endOffset;
 
         state.updateFollowerState(
             currentTimeMs,
@@ -886,6 +904,25 @@ public List<ReplicaKey> nonLeaderVotersByDescendingFetchOffset() {
             .collect(Collectors.toList());
     }
 
+    private void updateCommittedVoter(LogOffsetMetadata highWatermark) {
+        Map<Integer, ReplicaState> newCommittedVoterStates = new HashMap<>();
+
+        // Try to retrieve the voter set at the given high watermark offset;
+        // if unavailable, fall back to the static voter set.
+        // Note: highWatermark is the offset will be written so it is exclusive.
+        partitionState.voterSetAtOffset(highWatermark.offset() - 1).ifPresentOrElse(voterSet -> {
+            for (VoterSet.VoterNode voterNode : voterSet.voterNodes()) {
+                newCommittedVoterStates.put(voterNode.voterKey().id(), getOrCreateVoterReplicaState(voterNode));
+            }
+        }, () -> {
+            for (VoterSet.VoterNode voterNode : partitionState.staticVoterSet().voterNodes()) {
+                newCommittedVoterStates.put(voterNode.voterKey().id(), getOrCreateVoterReplicaState(voterNode));
+            }
+        });
+        log.debug("Updating committed voter at high watermark={} and committedVoter={}", highWatermark, newCommittedVoterStates);
+        committedVoterStates = newCommittedVoterStates;
+    }
+
     private Stream<ReplicaState> followersByDescendingFetchOffset() {
         return voterStates
             .values()
@@ -910,7 +947,7 @@ public long epochStartOffset() {
         return epochStartOffset;
     }
 
-    private ReplicaState getOrCreateReplicaState(ReplicaKey replicaKey) {
+    private ReplicaState getOrCreateObserverReplicaState(ReplicaKey replicaKey) {
         ReplicaState state = voterStates.get(replicaKey.id());
         if (state == null || !state.matchesKey(replicaKey)) {
             observerStates.putIfAbsent(replicaKey, new ReplicaState(replicaKey, false, Endpoints.empty()));
@@ -920,6 +957,10 @@ private ReplicaState getOrCreateReplicaState(ReplicaKey replicaKey) {
         return state;
     }
 
+    public ReplicaState getOrCreateVoterReplicaState(VoterSet.VoterNode voterNode) {
+        return getReplicaState(voterNode.voterKey()).orElse(new ReplicaState(voterNode.voterKey(), false, voterNode.listeners()));
+    }
+
     public Optional<ReplicaState> getReplicaState(ReplicaKey replicaKey) {
         ReplicaState state = voterStates.get(replicaKey.id());
         if (state == null || !state.matchesKey(replicaKey)) {
@@ -1128,12 +1169,13 @@ public boolean canGrantVote(ReplicaKey replicaKey, boolean isLogUpToDate, boolea
     @Override
     public String toString() {
         return String.format(
-            "Leader(localVoterNode=%s, epoch=%d, epochStartOffset=%d, highWatermark=%s, voterStates=%s)",
+            "Leader(localVoterNode=%s, epoch=%d, epochStartOffset=%d, highWatermark=%s, voterStates=%s, committedVoterState=%s)",
             localVoterNode,
             epoch,
             epochStartOffset,
             highWatermark,
-            voterStates
+            voterStates,
+            committedVoterStates
         );
     }
 

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

@@ -719,7 +719,8 @@ public <T> LeaderState<T> transitionToLeader(long epochStartOffset, BatchAccumul
             accumulator,
             fetchTimeoutMs,
             logContext,
-            kafkaRaftMetrics
+            kafkaRaftMetrics,
+            partitionState
         );
 
         durableTransitionTo(state);

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

@@ -490,6 +490,7 @@ public static DescribeQuorumResponseData singletonDescribeQuorumResponse(
         int leaderEpoch,
         long highWatermark,
         Collection<LeaderState.ReplicaState> voters,
+        Collection<LeaderState.ReplicaState> committedVoters,
         Collection<LeaderState.ReplicaState> observers,
         long currentTimeMs
     ) {
@@ -507,6 +508,7 @@ public static DescribeQuorumResponseData singletonDescribeQuorumResponse(
                                     .setLeaderEpoch(leaderEpoch)
                                     .setHighWatermark(highWatermark)
                                     .setCurrentVoters(toReplicaStates(apiVersion, leaderId, voters, currentTimeMs))
+                                    .setCommittedVoters(toReplicaStates(apiVersion, leaderId, committedVoters, currentTimeMs))
                                     .setObservers(toReplicaStates(apiVersion, leaderId, observers, currentTimeMs))))));
         if (apiVersion >= 2) {
             DescribeQuorumResponseData.NodeCollection nodes = new DescribeQuorumResponseData.NodeCollection(voters.size());

raft/src/main/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachine.java

@@ -202,6 +202,10 @@ public Optional<VoterSet> voterSetAtOffset(long offset) {
         }
     }
 
+    public VoterSet staticVoterSet() {
+        return staticVoterSet;
+    }
+
     /**
      * Returns the finalized kraft version at a given offset.
      *