[GitHub] [kafka] hachikuji commented on a diff in pull request #12081: MINOR: Refactor `kafka.cluster.Replica`

[GitBox <gi...@apache.org>]

hachikuji commented on code in PR #12081:
URL: https://github.com/apache/kafka/pull/12081#discussion_r856597477


##########
core/src/main/scala/kafka/cluster/Partition.scala:
##########
@@ -885,15 +885,18 @@ class Partition(val topicPartition: TopicPartition,
    *
    * @return true if the HW was incremented, and false otherwise.
    */
-  private def maybeIncrementLeaderHW(leaderLog: UnifiedLog, curTime: Long = time.milliseconds): Boolean = {
+  private def maybeIncrementLeaderHW(leaderLog: UnifiedLog, currentTimeMs: Long = time.milliseconds): Boolean = {

Review Comment:
   Do we need the default argument for `currentTimeMs`? Looks like we have already computed the time in `makeLeader`.



##########
core/src/main/scala/kafka/cluster/Replica.scala:
##########
@@ -62,39 +94,85 @@ class Replica(val brokerId: Int, val topicPartition: TopicPartition) extends Log
    * fetch request is always smaller than the leader's LEO, which can happen if small produce requests are received at
    * high frequency.
    */
-  def updateFetchState(followerFetchOffsetMetadata: LogOffsetMetadata,
-                       followerStartOffset: Long,
-                       followerFetchTimeMs: Long,
-                       leaderEndOffset: Long): Unit = {
-    if (followerFetchOffsetMetadata.messageOffset >= leaderEndOffset)
-      _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, followerFetchTimeMs)
-    else if (followerFetchOffsetMetadata.messageOffset >= lastFetchLeaderLogEndOffset)
-      _lastCaughtUpTimeMs = math.max(_lastCaughtUpTimeMs, lastFetchTimeMs)
-
-    _logStartOffset = followerStartOffset
-    _logEndOffsetMetadata = followerFetchOffsetMetadata
-    lastFetchLeaderLogEndOffset = leaderEndOffset
-    lastFetchTimeMs = followerFetchTimeMs
+  def updateFetchState(
+    followerFetchOffsetMetadata: LogOffsetMetadata,
+    followerStartOffset: Long,
+    followerFetchTimeMs: Long,
+    leaderEndOffset: Long
+  ): Unit = {
+    replicaState.updateAndGet { currentReplicaState =>
+      val lastCaughtUpTime = if (followerFetchOffsetMetadata.messageOffset >= leaderEndOffset) {
+        math.max(currentReplicaState.lastCaughtUpTimeMs, followerFetchTimeMs)
+      } else if (followerFetchOffsetMetadata.messageOffset >= currentReplicaState.lastFetchLeaderLogEndOffset) {
+        math.max(currentReplicaState.lastCaughtUpTimeMs, currentReplicaState.lastFetchTimeMs)
+      } else {
+        currentReplicaState.lastCaughtUpTimeMs
+      }
+
+      ReplicaState(
+        logStartOffset = followerStartOffset,
+        logEndOffsetMetadata = followerFetchOffsetMetadata,
+        lastFetchLeaderLogEndOffset = math.max(leaderEndOffset, currentReplicaState.lastFetchLeaderLogEndOffset),
+        lastFetchTimeMs = followerFetchTimeMs,
+        lastCaughtUpTimeMs = lastCaughtUpTime
+      )
+    }
   }
 
-  def resetLastCaughtUpTime(curLeaderLogEndOffset: Long, curTimeMs: Long, lastCaughtUpTimeMs: Long): Unit = {
-    lastFetchLeaderLogEndOffset = curLeaderLogEndOffset
-    lastFetchTimeMs = curTimeMs
-    _lastCaughtUpTimeMs = lastCaughtUpTimeMs
+  /**
+   * When the leader is elected or re-elected, the state of the follower is reinitialized
+   * accordingly.
+   */
+  def resetReplicaState(
+    currentTimeMs: Long,
+    leaderEndOffset: Long,
+    isNewLeader: Boolean,
+    isFollowerInSync: Boolean
+  ): Unit = {
+    replicaState.updateAndGet { currentReplicaState =>
+      // When the leader is elected or re-elected, the follower's last caught up time
+      // is set to the current time if the follower is in the ISR, else to 0. The later

Review Comment:
   nit: The latter?



##########
core/src/test/scala/unit/kafka/cluster/LogTest.scala:
##########
@@ -0,0 +1,132 @@
+/*

Review Comment:
   I'm ok with that, but it might not be as much work as you imagine: `UnifiedLogTest` has a pretty minimalistic setup.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org