This is an automated email from the ASF dual-hosted git repository.
divijv pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 72503904e8e MINOR: Log lastCaughtUpTime on ISR shrinkage (#13187)
72503904e8e is described below
commit 72503904e8e594a20ad7efe60f223390fb312335
Author: Joseph (Ting-Chou) Lin <[email protected]>
AuthorDate: Wed Jun 21 01:15:50 2023 -0700
MINOR: Log lastCaughtUpTime on ISR shrinkage (#13187)
Reviewers: Divij Vaidya <[email protected]>
---
core/src/main/scala/kafka/cluster/Partition.scala | 10 +++++++---
1 file changed, 7 insertions(+), 3 deletions(-)
diff --git a/core/src/main/scala/kafka/cluster/Partition.scala
b/core/src/main/scala/kafka/cluster/Partition.scala
index a84b5463720..4913a787ef3 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -1170,10 +1170,14 @@ class Partition(val topicPartition: TopicPartition,
partitionState match {
case currentState: CommittedPartitionState if
outOfSyncReplicaIds.nonEmpty =>
val outOfSyncReplicaLog = outOfSyncReplicaIds.map { replicaId =>
- val logEndOffsetMessage = getReplica(replicaId)
- .map(_.stateSnapshot.logEndOffset.toString)
+ val replicaStateSnapshot =
getReplica(replicaId).map(_.stateSnapshot)
+ val logEndOffsetMessage = replicaStateSnapshot
+ .map(_.logEndOffset.toString)
.getOrElse("unknown")
- s"(brokerId: $replicaId, endOffset: $logEndOffsetMessage)"
+ val lastCaughtUpTimeMessage = replicaStateSnapshot
+ .map(_.lastCaughtUpTimeMs.toString)
+ .getOrElse("unknown")
+ s"(brokerId: $replicaId, endOffset: $logEndOffsetMessage,
lastCaughtUpTimeMs: $lastCaughtUpTimeMessage)"
}.mkString(" ")
val newIsrLog = (partitionState.isr --
outOfSyncReplicaIds).mkString(",")
info(s"Shrinking ISR from ${partitionState.isr.mkString(",")} to
$newIsrLog. " +
