KAFKA-13888: Addition of Information in DescribeQuorumResponse

core/src/main/scala/kafka/network/RequestChannel.scala

@@ -121,10 +121,17 @@ object RequestChannel extends Logging {
 
     def isForwarded: Boolean = envelope.isDefined
 
+    private def shouldReturnNotController(response: AbstractResponse): Boolean = {
+      response match {
+        case describeQuorumResponse: DescribeQuorumResponse => response.errorCounts.containsKey(Errors.NOT_LEADER_OR_FOLLOWER)
+        case _ => response.errorCounts.containsKey(Errors.NOT_CONTROLLER)
+      }
+    }
+
     def buildResponseSend(abstractResponse: AbstractResponse): Send = {
       envelope match {
         case Some(request) =>
-          val envelopeResponse = if (abstractResponse.errorCounts().containsKey(Errors.NOT_CONTROLLER)) {
+          val envelopeResponse = if (shouldReturnNotController(abstractResponse)) {
             // Since it's a NOT_CONTROLLER error response, we need to make envelope response with NOT_CONTROLLER error
             // to notify the requester (i.e. BrokerToControllerRequestThread) to update active controller
             new EnvelopeResponse(new EnvelopeResponseData()

core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala

@@ -21,7 +21,7 @@ import kafka.network.SocketServer
 import kafka.server.IntegrationTestUtils.connectAndReceive
 import kafka.testkit.{BrokerNode, KafkaClusterTestKit, TestKitNodes}
 import kafka.utils.TestUtils
-import org.apache.kafka.clients.admin.{Admin, AlterConfigOp, Config, ConfigEntry, NewPartitionReassignment, NewTopic}
+import org.apache.kafka.clients.admin.{Admin, AdminClientConfig, AlterConfigOp, Config, ConfigEntry, DescribeMetadataQuorumOptions, NewPartitionReassignment, NewTopic}
 import org.apache.kafka.common.{TopicPartition, TopicPartitionInfo}
 import org.apache.kafka.common.message.DescribeClusterRequestData
 import org.apache.kafka.common.network.ListenerName
@@ -32,7 +32,7 @@ import org.junit.jupiter.api.Assertions._
 import org.junit.jupiter.api.{Tag, Test, Timeout}
 
 import java.util
-import java.util.{Arrays, Collections, Optional}
+import java.util.{Arrays, Collections, Optional, OptionalLong, Properties}
 import org.apache.kafka.clients.admin.AlterConfigOp.OpType
 import org.apache.kafka.common.config.ConfigResource
 import org.apache.kafka.common.config.ConfigResource.Type
@@ -778,4 +778,55 @@ class KRaftClusterTest {
       cluster.close()
     }
   }
+  def createAdminClient(cluster: KafkaClusterTestKit): Admin = {
+    var props: Properties = null
+    props = cluster.clientProperties()
+    props.put(AdminClientConfig.CLIENT_ID_CONFIG, this.getClass.getName)
+    Admin.create(props)
+  }
+
+  @Test
+  def testDescribeQuorumRequestToBrokers() : Unit = {
+    val cluster = new KafkaClusterTestKit.Builder(
+      new TestKitNodes.Builder().
+        setNumBrokerNodes(4).
+        setNumControllerNodes(3).build()).build()
+    try {
+      cluster.format
+      cluster.startup
+      for (i <- 0 to 3) {
+        TestUtils.waitUntilTrue(() => cluster.brokers.get(i).brokerState == BrokerState.RUNNING,
+          "Broker Never started up")
+      }
+      val admin = createAdminClient(cluster)
+      try {
+        val quorumState = admin.describeMetadataQuorum(new DescribeMetadataQuorumOptions)
+        val quorumInfo = quorumState.quorumInfo.get()
+
+        assertEquals(cluster.controllers.asScala.keySet, quorumInfo.voters.asScala.map(_.replicaId).toSet)
+        assertTrue(cluster.controllers.asScala.keySet.contains(quorumInfo.leaderId),
+          s"Leader ID ${quorumInfo.leaderId} was not a controller ID.")
+
+        quorumInfo.voters.forEach { voter =>
+          assertTrue(0 < voter.logEndOffset,
+            s"logEndOffset for voter with ID ${voter.replicaId} was ${voter.logEndOffset}")
+          assertNotEquals(OptionalLong.empty(), voter.lastFetchTimeMs)
+          assertNotEquals(OptionalLong.empty(), voter.lastCaughtUpTimeMs)
+        }
+
+        assertEquals(cluster.brokers.asScala.keySet, quorumInfo.observers.asScala.map(_.replicaId).toSet)
+        quorumInfo.observers.forEach { observer =>
+          assertTrue(0 < observer.logEndOffset,
+            s"logEndOffset for observer with ID ${observer.replicaId} was ${observer.logEndOffset}")
+          assertNotEquals(OptionalLong.empty(), observer.lastFetchTimeMs)
+          assertNotEquals(OptionalLong.empty(), observer.lastCaughtUpTimeMs)
+        }
+      } finally {
+        admin.close()
+      }
+    } finally {
+      cluster.close()
+    }
+  }
+
 }

core/src/test/scala/unit/kafka/server/DescribeQuorumRequestTest.scala

@@ -87,8 +87,13 @@ class DescribeQuorumRequestTest(cluster: ClusterInstance) {
 
       (voterData ++ observerData).foreach { state =>
         assertTrue(0 < state.logEndOffset)
-        assertEquals(-1, state.lastFetchTimestamp)
-        assertEquals(-1, state.lastCaughtUpTimestamp)
+        if (version == 0) {
+          assertEquals(-1, state.lastFetchTimestamp)
+          assertEquals(-1, state.lastCaughtUpTimestamp)
+        } else {
+          assertNotEquals(-1, state.lastFetchTimestamp)
+          assertNotEquals(-1, state.lastCaughtUpTimestamp)
+        }
       }
     }
   }

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

@@ -25,7 +25,6 @@
 import org.apache.kafka.common.message.BeginQuorumEpochResponseData;
 import org.apache.kafka.common.message.DescribeQuorumRequestData;
 import org.apache.kafka.common.message.DescribeQuorumResponseData;
-import org.apache.kafka.common.message.DescribeQuorumResponseData.ReplicaState;
 import org.apache.kafka.common.message.EndQuorumEpochRequestData;
 import org.apache.kafka.common.message.EndQuorumEpochResponseData;
 import org.apache.kafka.common.message.FetchRequestData;
@@ -95,7 +94,6 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
-import java.util.stream.Collectors;
 
 import static java.util.concurrent.CompletableFuture.completedFuture;
 import static org.apache.kafka.raft.RaftUtil.hasValidTopicPartition;
@@ -1016,7 +1014,7 @@ private FetchResponseData tryCompleteFetchRequest(
             if (validOffsetAndEpoch.kind() == ValidOffsetAndEpoch.Kind.VALID) {
                 LogFetchInfo info = log.read(fetchOffset, Isolation.UNCOMMITTED);
 
-                if (state.updateReplicaState(replicaId, currentTimeMs, info.startOffsetMetadata)) {
+                if (state.updateReplicaState(replicaId, currentTimeMs, info.startOffsetMetadata, log.endOffset().offset)) {
                     onUpdateLeaderHighWatermark(state, currentTimeMs);
                 }
 
@@ -1182,8 +1180,8 @@ private DescribeQuorumResponseData handleDescribeQuorumRequest(
             leaderState.localId(),
             leaderState.epoch(),
             leaderState.highWatermark().isPresent() ? leaderState.highWatermark().get().offset : -1,
-            convertToReplicaStates(leaderState.getVoterEndOffsets()),
-            convertToReplicaStates(leaderState.getObserverStates(currentTimeMs))
+            leaderState.quorumResponseVoterStates(currentTimeMs),
+            leaderState.quorumResponseObserverStates(currentTimeMs)
         );
     }
 
@@ -1421,14 +1419,6 @@ private boolean handleFetchSnapshotResponse(
         return true;
     }
 
-    List<ReplicaState> convertToReplicaStates(Map<Integer, Long> replicaEndOffsets) {
-        return replicaEndOffsets.entrySet().stream()
-                   .map(entry -> new ReplicaState()
-                                     .setReplicaId(entry.getKey())
-                                     .setLogEndOffset(entry.getValue()))
-                   .collect(Collectors.toList());
-    }
-
     private boolean hasConsistentLeader(int epoch, OptionalInt leaderId) {
         // Only elected leaders are sent in the request/response header, so if we have an elected
         // leaderId, it should be consistent with what is in the message.

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.raft;
 
+import org.apache.kafka.common.message.DescribeQuorumResponseData;
 import org.apache.kafka.common.utils.LogContext;
 import org.apache.kafka.raft.internals.BatchAccumulator;
 import org.slf4j.Logger;
@@ -25,6 +26,7 @@
 import org.apache.kafka.common.record.ControlRecordUtils;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -205,10 +207,10 @@ private boolean updateHighWatermark() {
     /**
      * Update the local replica state.
      *
-     * See {@link #updateReplicaState(int, long, LogOffsetMetadata)}
+     * See {@link #updateReplicaState(int, long, LogOffsetMetadata, long)}
      */
     public boolean updateLocalState(long fetchTimestamp, LogOffsetMetadata logOffsetMetadata) {
-        return updateReplicaState(localId, fetchTimestamp, logOffsetMetadata);
+        return updateReplicaState(localId, fetchTimestamp, logOffsetMetadata, logOffsetMetadata.offset);
     }
 
     /**
@@ -217,19 +219,34 @@ public boolean updateLocalState(long fetchTimestamp, LogOffsetMetadata logOffset
      * @param replicaId replica id
      * @param fetchTimestamp fetch timestamp
      * @param logOffsetMetadata new log offset and metadata
+     * @param leaderLogEndOffset current log end offset of the leader
      * @return true if the high watermark is updated too
      */
-    public boolean updateReplicaState(int replicaId,
-                                      long fetchTimestamp,
-                                      LogOffsetMetadata logOffsetMetadata) {
+    public boolean updateReplicaState(
+        int replicaId,
+        long fetchTimestamp,
+        LogOffsetMetadata logOffsetMetadata,
+        long leaderLogEndOffset
+    ) {
         // Ignore fetches from negative replica id, as it indicates
         // the fetch is from non-replica. For example, a consumer.
         if (replicaId < 0) {
             return false;
         }
 
         ReplicaState state = getReplicaState(replicaId);
-        state.updateFetchTimestamp(fetchTimestamp);
+
+        // Only proceed with updating the states if the offset update is valid
+        verifyEndOffsetUpdate(state, logOffsetMetadata);
+
+        // Update the Last CaughtUp Time
+        if (logOffsetMetadata.offset >= leaderLogEndOffset) {
+            state.updateLastCaughtUpTimestamp(fetchTimestamp);
+        } else if (logOffsetMetadata.offset >= state.lastFetchLeaderLogEndOffset.orElse(-1L)) {
+            state.updateLastCaughtUpTimestamp(state.lastFetchTimestamp.orElse(-1L));
+        }
+
+        state.updateFetchTimestamp(fetchTimestamp, leaderLogEndOffset);
         return updateEndOffset(state, logOffsetMetadata);
     }
 
@@ -246,8 +263,10 @@ private List<ReplicaState> followersByDescendingFetchOffset() {
             .collect(Collectors.toList());
     }
 
-    private boolean updateEndOffset(ReplicaState state,
-                                    LogOffsetMetadata endOffsetMetadata) {
+    private void verifyEndOffsetUpdate(
+        ReplicaState state,
+        LogOffsetMetadata endOffsetMetadata
+    ) {
         state.endOffset.ifPresent(currentEndOffset -> {
             if (currentEndOffset.offset > endOffsetMetadata.offset) {
                 if (state.nodeId == localId) {
@@ -259,7 +278,11 @@ private boolean updateEndOffset(ReplicaState state,
                 }
             }
         });
-
+    }
+    private boolean updateEndOffset(
+        ReplicaState state,
+        LogOffsetMetadata endOffsetMetadata
+    ) {
         state.endOffset = Optional.of(endOffsetMetadata);
         state.hasAcknowledgedLeader = true;
         return isVoter(state.nodeId) && updateHighWatermark();
@@ -290,22 +313,36 @@ private ReplicaState getReplicaState(int remoteNodeId) {
         return state;
     }
 
-    Map<Integer, Long> getVoterEndOffsets() {
-        return getReplicaEndOffsets(voterStates);
+    List<DescribeQuorumResponseData.ReplicaState> quorumResponseVoterStates(long currentTimeMs) {
+        return quorumResponseReplicaStates(voterStates.values(), localId, currentTimeMs);
     }
 
-    Map<Integer, Long> getObserverStates(final long currentTimeMs) {
+    List<DescribeQuorumResponseData.ReplicaState> quorumResponseObserverStates(long currentTimeMs) {
         clearInactiveObservers(currentTimeMs);
-        return getReplicaEndOffsets(observerStates);
+        return quorumResponseReplicaStates(observerStates.values(), localId, currentTimeMs);
     }
 
-    private static <R extends ReplicaState> Map<Integer, Long> getReplicaEndOffsets(
-        Map<Integer, R> replicaStates) {
-        return replicaStates.entrySet().stream()
-                   .collect(Collectors.toMap(Map.Entry::getKey,
-                       e -> e.getValue().endOffset.map(
-                           logOffsetMetadata -> logOffsetMetadata.offset).orElse(-1L))
-                   );
+    private static  List<DescribeQuorumResponseData.ReplicaState> quorumResponseReplicaStates(
+        Collection<ReplicaState> state,
+        int leaderId,
+        long currentTimeMs
+    ) {
+        return state.stream().map(s -> {
+            final long lastCaughtUpTimestamp;
+            final long lastFetchTimestamp;
+            if (s.nodeId == leaderId) {
+                lastCaughtUpTimestamp = currentTimeMs;
+                lastFetchTimestamp = currentTimeMs;
+            } else {
+                lastCaughtUpTimestamp = s.lastCaughtUpTimestamp.orElse(-1);
+                lastFetchTimestamp = s.lastFetchTimestamp.orElse(-1);
+            }
+            return new DescribeQuorumResponseData.ReplicaState()
+                    .setReplicaId(s.nodeId)
+                    .setLogEndOffset(s.endOffset.map(md -> md.offset).orElse(-1L))
+                    .setLastCaughtUpTimestamp(lastCaughtUpTimestamp)
+                    .setLastFetchTimestamp(lastFetchTimestamp);
+        }).collect(Collectors.toList());
     }
 
     private void clearInactiveObservers(final long currentTimeMs) {
@@ -323,19 +360,30 @@ private static class ReplicaState implements Comparable<ReplicaState> {
         final int nodeId;
         Optional<LogOffsetMetadata> endOffset;
         OptionalLong lastFetchTimestamp;
+        OptionalLong lastFetchLeaderLogEndOffset;
+        OptionalLong lastCaughtUpTimestamp;
         boolean hasAcknowledgedLeader;
 
         public ReplicaState(int nodeId, boolean hasAcknowledgedLeader) {
             this.nodeId = nodeId;
             this.endOffset = Optional.empty();
             this.lastFetchTimestamp = OptionalLong.empty();
+            this.lastFetchLeaderLogEndOffset = OptionalLong.empty();
+            this.lastCaughtUpTimestamp = OptionalLong.empty();
             this.hasAcknowledgedLeader = hasAcknowledgedLeader;
         }
 
-        void updateFetchTimestamp(long currentFetchTimeMs) {
+        void updateFetchTimestamp(long currentFetchTimeMs, long leaderLogEndOffset) {
             // To be resilient to system time shifts we do not strictly
             // require the timestamp be monotonically increasing.
             lastFetchTimestamp = OptionalLong.of(Math.max(lastFetchTimestamp.orElse(-1L), currentFetchTimeMs));
+            lastFetchLeaderLogEndOffset = OptionalLong.of(leaderLogEndOffset);
+        }
+
+        void updateLastCaughtUpTimestamp(long lastCaughtUpTime) {
+            // This value relies on the fetch timestamp which does not
+            // require monotonicity
+            lastCaughtUpTimestamp = OptionalLong.of(Math.max(lastCaughtUpTimestamp.orElse(-1L), lastCaughtUpTime));
         }
 
         @Override
@@ -353,10 +401,12 @@ else if (!that.endOffset.isPresent())
         @Override
         public String toString() {
             return String.format(
-                "ReplicaState(nodeId=%d, endOffset=%s, lastFetchTimestamp=%s, hasAcknowledgedLeader=%s)",
+                "ReplicaState(nodeId=%d, endOffset=%s, lastFetchTimestamp=%s, " +
+                        "lastCaughtUpTimestamp=%s, hasAcknowledgedLeader=%s)",
                 nodeId,
                 endOffset,
                 lastFetchTimestamp,
+                lastCaughtUpTimestamp,
                 hasAcknowledgedLeader 
             );
         }