From 2239b5630dff27c6c94856918ef1643fedd22cc1 Mon Sep 17 00:00:00 2001
From: Dhruvil Shah <dhruvil@confluent.io>
Date: Fri, 18 Jan 2019 11:16:32 -0800
Subject: [PATCH 1/3] KAFKA-7838: Log leader and follower end offsets when
 shrinking ISR

---
 core/src/main/scala/kafka/cluster/Partition.scala | 12 +++++++++---
 1 file changed, 9 insertions(+), 3 deletions(-)

diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
index ca3abbbc97348..9373987c8ebe8 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -634,16 +634,22 @@ class Partition(val topicPartition: TopicPartition,
       leaderReplicaIfLocal match {
         case Some(leaderReplica) =>
           val outOfSyncReplicas = getOutOfSyncReplicas(leaderReplica, replicaMaxLagTimeMs)
-          if(outOfSyncReplicas.nonEmpty) {
+          if (outOfSyncReplicas.nonEmpty) {
             val newInSyncReplicas = inSyncReplicas -- outOfSyncReplicas
             assert(newInSyncReplicas.nonEmpty)
             info("Shrinking ISR from %s to %s".format(inSyncReplicas.map(_.brokerId).mkString(","),
               newInSyncReplicas.map(_.brokerId).mkString(",")))
+            debug(s"Leader (hwm: ${leaderReplica.highWatermark.messageOffset}, endOffset: ${leaderReplica.logEndOffset.messageOffset})")
+            debug(s"Out of sync replicas: " +
+              s"${outOfSyncReplicas.map { replica =>
+                s"(brokerId: ${replica.brokerId}, endOffset: ${replica.logEndOffset.messageOffset})"
+              }.mkString(" ")}")
+
             // update ISR in zk and in cache
             updateIsr(newInSyncReplicas)
-            // we may need to increment high watermark since ISR could be down to 1
-
             replicaManager.isrShrinkRate.mark()
+
+            // we may need to increment high watermark since ISR could be down to 1
             maybeIncrementLeaderHW(leaderReplica)
           } else {
             false

From 74ecbcaea8be9158fe014efbabb43963f3598f6c Mon Sep 17 00:00:00 2001
From: Dhruvil Shah <dhruvil@confluent.io>
Date: Wed, 23 Jan 2019 22:45:06 -0800
Subject: [PATCH 2/3] hwm --> highWatermark

---
 core/src/main/scala/kafka/cluster/Partition.scala | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
index 9373987c8ebe8..60fd28adb0c7b 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -639,7 +639,7 @@ class Partition(val topicPartition: TopicPartition,
             assert(newInSyncReplicas.nonEmpty)
             info("Shrinking ISR from %s to %s".format(inSyncReplicas.map(_.brokerId).mkString(","),
               newInSyncReplicas.map(_.brokerId).mkString(",")))
-            debug(s"Leader (hwm: ${leaderReplica.highWatermark.messageOffset}, endOffset: ${leaderReplica.logEndOffset.messageOffset})")
+            debug(s"Leader (highWatermark: ${leaderReplica.highWatermark.messageOffset}, endOffset: ${leaderReplica.logEndOffset.messageOffset})")
             debug(s"Out of sync replicas: " +
               s"${outOfSyncReplicas.map { replica =>
                 s"(brokerId: ${replica.brokerId}, endOffset: ${replica.logEndOffset.messageOffset})"

From 11fbfbee0ffae8aae79797541d56a680a20eaf7f Mon Sep 17 00:00:00 2001
From: Dhruvil Shah <dhruvil@confluent.io>
Date: Thu, 24 Jan 2019 16:58:46 -0800
Subject: [PATCH 3/3] Address review comment

---
 .../main/scala/kafka/cluster/Partition.scala    | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/core/src/main/scala/kafka/cluster/Partition.scala b/core/src/main/scala/kafka/cluster/Partition.scala
index 60fd28adb0c7b..e731111ab4836 100755
--- a/core/src/main/scala/kafka/cluster/Partition.scala
+++ b/core/src/main/scala/kafka/cluster/Partition.scala
@@ -637,13 +637,16 @@ class Partition(val topicPartition: TopicPartition,
           if (outOfSyncReplicas.nonEmpty) {
             val newInSyncReplicas = inSyncReplicas -- outOfSyncReplicas
             assert(newInSyncReplicas.nonEmpty)
-            info("Shrinking ISR from %s to %s".format(inSyncReplicas.map(_.brokerId).mkString(","),
-              newInSyncReplicas.map(_.brokerId).mkString(",")))
-            debug(s"Leader (highWatermark: ${leaderReplica.highWatermark.messageOffset}, endOffset: ${leaderReplica.logEndOffset.messageOffset})")
-            debug(s"Out of sync replicas: " +
-              s"${outOfSyncReplicas.map { replica =>
-                s"(brokerId: ${replica.brokerId}, endOffset: ${replica.logEndOffset.messageOffset})"
-              }.mkString(" ")}")
+            info("Shrinking ISR from %s to %s. Leader: (highWatermark: %d, endOffset: %d). Out of sync replicas: %s."
+              .format(inSyncReplicas.map(_.brokerId).mkString(","),
+                newInSyncReplicas.map(_.brokerId).mkString(","),
+                leaderReplica.highWatermark.messageOffset,
+                leaderReplica.logEndOffset.messageOffset,
+                outOfSyncReplicas.map { replica =>
+                  s"(brokerId: ${replica.brokerId}, endOffset: ${replica.logEndOffset.messageOffset})"
+                }.mkString(" ")
+              )
+            )
 
             // update ISR in zk and in cache
             updateIsr(newInSyncReplicas)