KAFKA-16524: Metrics for KIP-853

core/src/main/scala/kafka/raft/DefaultExternalKRaftMetrics.scala

@@ -22,12 +22,12 @@ import org.apache.kafka.raft.ExternalKRaftMetrics
 import org.apache.kafka.server.metrics.BrokerServerMetrics
 
 class DefaultExternalKRaftMetrics(
-  val brokerServerMetricsOpt: Option[BrokerServerMetrics],
-  val controllerMetadataMetricsOpt: Option[ControllerMetadataMetrics]
+  val brokerServerMetrics: Option[BrokerServerMetrics],
+  val controllerMetadataMetrics: Option[ControllerMetadataMetrics]
 ) extends ExternalKRaftMetrics {
 
   override def setIgnoredStaticVoters(ignoredStaticVoters: Boolean): Unit = {
-    brokerServerMetricsOpt.foreach(metrics => metrics.setIgnoredStaticVoters(ignoredStaticVoters))
-    controllerMetadataMetricsOpt.foreach(metrics => metrics.setIgnoredStaticVoters(ignoredStaticVoters))
+    brokerServerMetrics.foreach(metrics => metrics.setIgnoredStaticVoters(ignoredStaticVoters))
+    controllerMetadataMetrics.foreach(metrics => metrics.setIgnoredStaticVoters(ignoredStaticVoters))
   }
 }

core/src/test/scala/kafka/raft/DefaultExternalKRaftMetricsTest.scala

@@ -30,7 +30,7 @@ final class DefaultExternalKRaftMetricsTest {
   def testDefaultExternalKRaftMetrics(): Unit = {
     val brokerServerMetrics = new BrokerServerMetrics(new Metrics())
     val controllerMetadataMetrics = new ControllerMetadataMetrics(Optional.of(new MetricsRegistry()))
-    var metrics = new DefaultExternalKRaftMetrics(
+    val metrics = new DefaultExternalKRaftMetrics(
       Option(brokerServerMetrics),
       Option(controllerMetadataMetrics)
     )
@@ -47,8 +47,11 @@ final class DefaultExternalKRaftMetricsTest {
 
     assertFalse(brokerServerMetrics.ignoredStaticVoters())
     assertFalse(controllerMetadataMetrics.ignoredStaticVoters())
+  }
 
-    metrics = new DefaultExternalKRaftMetrics(None, None)
+  @Test
+  def testEmptyDefaultExternalKRaftMetrics(): Unit = {
+    val metrics = new DefaultExternalKRaftMetrics(None, None)
     metrics.setIgnoredStaticVoters(true)
   }
 }

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

@@ -125,6 +125,8 @@ protected LeaderState(
         this.voterSetAtEpochStart =  voterSetAtEpochStart;
         this.offsetOfVotersAtEpochStart = offsetOfVotersAtEpochStart;
         this.kraftVersionAtEpochStart = kraftVersionAtEpochStart;
+
+        kafkaRaftMetrics.addLeaderMetrics();
         this.kafkaRaftMetrics = kafkaRaftMetrics;
     }
 

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

@@ -730,7 +730,6 @@ public <T> LeaderState<T> transitionToLeader(long epochStartOffset, BatchAccumul
             logContext,
             kafkaRaftMetrics
         );
-        kafkaRaftMetrics.addLeaderMetrics();
 
         durableTransitionTo(state);
         return state;

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

@@ -73,7 +73,7 @@ public final class KRaftControlRecordStateMachine {
     private volatile long nextOffset = STARTING_NEXT_OFFSET;
     private final KafkaRaftMetrics kafkaRaftMetrics;
     private final ExternalKRaftMetrics externalKRaftMetrics;
-    private final Optional<VoterSet> staticVoterSet;
+    private final VoterSet staticVoterSet;
 
     /**
      * Constructs an internal log listener
@@ -103,9 +103,11 @@ public KRaftControlRecordStateMachine(
         this.logger = logContext.logger(this.getClass());
         this.kafkaRaftMetrics = kafkaRaftMetrics;
         this.externalKRaftMetrics = externalKRaftMetrics;
-        this.staticVoterSet = staticVoterSet.size() > 0 ? Optional.of(staticVoterSet) : Optional.empty();
+        this.staticVoterSet = staticVoterSet;
 
-        this.staticVoterSet.ifPresent(voters -> kafkaRaftMetrics.updateNumVoters(voters.size()));
+        if (staticVoterSet.size() > 0) {
+            kafkaRaftMetrics.updateNumVoters(staticVoterSet.size());
+        }
     }
 
     /**
@@ -130,7 +132,7 @@ public void truncateNewEntries(long endOffset) {
         }
 
         kafkaRaftMetrics.updateNumVoters(voterSetHistory.lastValue().size());
-        if (staticVoterSet.isPresent() && voterSetHistory.lastEntry().isEmpty()) {
+        if (staticVoterSet.size() > 0 && voterSetHistory.lastEntry().isEmpty()) {
             externalKRaftMetrics.setIgnoredStaticVoters(false);
         }
     }
@@ -298,7 +300,7 @@ private void handleBatch(Batch<?> batch, OptionalLong overrideOffset) {
                 case KRAFT_VOTERS:
                     VoterSet voters = VoterSet.fromVotersRecord((VotersRecord) record.message());
                     kafkaRaftMetrics.updateNumVoters(voters.size());
-                    if (staticVoterSet.isPresent()) {
+                    if (staticVoterSet.size() > 0) {
                         externalKRaftMetrics.setIgnoredStaticVoters(true);
                     }
                     logger.info("Latest set of voters is {} at offset {}", voters, currentOffset);

raft/src/test/java/org/apache/kafka/raft/internals/KRaftControlRecordStateMachineTest.java

@@ -33,8 +33,8 @@
 import org.apache.kafka.server.common.serialization.RecordSerde;
 import org.apache.kafka.snapshot.RecordsSnapshotWriter;
 
-import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
 
 import java.util.Optional;
 import java.util.stream.IntStream;