MINOR: Improve raft log4j messages a bit (#13553)

Reviewers: Colin P. McCabe <cmccabe@apache.org>

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

@@ -192,17 +192,26 @@ final class KafkaMetadataLog private (
   }
 
   override def updateHighWatermark(offsetMetadata: LogOffsetMetadata): Unit = {
-    offsetMetadata.metadata.asScala match {
+    // This API returns the new high watermark, which may be different from the passed offset
-      case Some(segmentPosition: SegmentPosition) => log.updateHighWatermark(
+    val logHighWatermark = offsetMetadata.metadata.asScala match {
+      case Some(segmentPosition: SegmentPosition) =>
+        log.updateHighWatermark(
           new internals.log.LogOffsetMetadata(
             offsetMetadata.offset,
             segmentPosition.baseOffset,
-          segmentPosition.relativePosition)
+            segmentPosition.relativePosition
+          )
         )
       case _ =>
-        // FIXME: This API returns the new high watermark, which may be different from the passed offset
         log.updateHighWatermark(offsetMetadata.offset)
     }
+
+    // Temporary log message until we fix KAFKA-14825
+    if (logHighWatermark != offsetMetadata.offset) {
+      warn(
+        s"Log's high watermark ($logHighWatermark) is different from the local replica's high watermark ($offsetMetadata)"
+      )
+    }
   }
 
   override def highWatermark: LogOffsetMetadata = {

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

@@ -1091,7 +1091,11 @@ public class KafkaRaftClient<T> implements RaftClient<T> {
                 });
 
                 long truncationOffset = log.truncateToEndOffset(divergingOffsetAndEpoch);
-                logger.info("Truncated to offset {} from Fetch response from leader {}", truncationOffset, quorum.leaderIdOrSentinel());
+                logger.info(
+                    "Truncated to offset {} from Fetch response from leader {}",
+                    truncationOffset,
+                    quorum.leaderIdOrSentinel()
+                );
             } else if (partitionResponse.snapshotId().epoch() >= 0 ||
                        partitionResponse.snapshotId().endOffset() >= 0) {
                 // The leader is asking us to fetch a snapshot
@@ -1120,6 +1124,11 @@ public class KafkaRaftClient<T> implements RaftClient<T> {
                     // since this snapshot is expected to reference offsets and epochs
                     // greater than the log end offset and high-watermark
                     state.setFetchingSnapshot(log.storeSnapshot(snapshotId));
+                    logger.info(
+                        "Fetching snapshot {} from Fetch response from leader {}",
+                        snapshotId,
+                        quorum.leaderIdOrSentinel()
+                    );
                 }
             } else {
                 Records records = FetchResponse.recordsOrFail(partitionResponse);
@@ -1407,6 +1416,14 @@ public class KafkaRaftClient<T> implements RaftClient<T> {
             state.setFetchingSnapshot(Optional.empty());
 
             if (log.truncateToLatestSnapshot()) {
+                logger.info(
+                    "Fully truncated the log at ({}, {}) after downloading snapshot {} from leader {}",
+                    log.endOffset(),
+                    log.lastFetchedEpoch(),
+                    snapshot.snapshotId(),
+                    quorum.leaderIdOrSentinel()
+                );
+
                 updateFollowerHighWatermark(state, OptionalLong.of(log.highWatermark().offset));
             } else {
                 throw new IllegalStateException(

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

@@ -212,7 +212,7 @@ public class LeaderState<T> implements EpochState {
         List<ReplicaState> followersByDescendingFetchOffset
     ) {
         if (oldHighWatermark.isPresent()) {
-            log.trace(
+            log.debug(
                 "High watermark set to {} from {} based on indexOfHw {} and voters {}",
                 newHighWatermark,
                 oldHighWatermark.get(),

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

@@ -209,7 +209,7 @@ public class QuorumState {
             );
         }
 
-        transitionTo(initialState);
+        durableTransitionTo(initialState);
     }
 
     public Set<Integer> remoteVoters() {
@@ -277,7 +277,8 @@ public class QuorumState {
         // The Resigned state is a soft state which does not need to be persisted.
         // A leader will always be re-initialized in this state.
         int epoch = state.epoch();
-        this.state = new ResignedState(
+        memoryTransitionTo(
+            new ResignedState(
                 time,
                 localIdOrThrow(),
                 epoch,
@@ -285,8 +286,8 @@ public class QuorumState {
                 randomElectionTimeoutMs(),
                 preferredSuccessors,
                 logContext
+            )
         );
-        log.info("Completed transition to {}", state);
     }
 
     /**
@@ -313,7 +314,7 @@ public class QuorumState {
             electionTimeoutMs = randomElectionTimeoutMs();
         }
 
-        transitionTo(new UnattachedState(
+        durableTransitionTo(new UnattachedState(
             time,
             epoch,
             voters,
@@ -356,7 +357,7 @@ public class QuorumState {
         // Note that we reset the election timeout after voting for a candidate because we
         // know that the candidate has at least as good of a chance of getting elected as us
 
-        transitionTo(new VotedState(
+        durableTransitionTo(new VotedState(
             time,
             epoch,
             candidateId,
@@ -392,7 +393,7 @@ public class QuorumState {
                 " and epoch=" + epoch + " from state " + state);
         }
 
-        transitionTo(new FollowerState(
+        durableTransitionTo(new FollowerState(
             time,
             epoch,
             leaderId,
@@ -416,7 +417,7 @@ public class QuorumState {
         int newEpoch = epoch() + 1;
         int electionTimeoutMs = randomElectionTimeoutMs();
 
-        transitionTo(new CandidateState(
+        durableTransitionTo(new CandidateState(
             time,
             localIdOrThrow(),
             newEpoch,
@@ -460,11 +461,11 @@ public class QuorumState {
             accumulator,
             logContext
         );
-        transitionTo(state);
+        durableTransitionTo(state);
         return state;
     }
 
-    private void transitionTo(EpochState state) {
+    private void durableTransitionTo(EpochState state) {
         if (this.state != null) {
             try {
                 this.state.close();
@@ -475,6 +476,10 @@ public class QuorumState {
         }
 
         this.store.writeElectionState(state.election());
+        memoryTransitionTo(state);
+    }
+
+    private void memoryTransitionTo(EpochState state) {
         EpochState from = this.state;
         this.state = state;
         log.info("Completed transition to {} from {}", state, from);