KAFKA-19996: Avoid premature ISR expansion under min ISR

core/src/main/scala/kafka/cluster/Partition.scala

@@ -867,9 +867,10 @@ class Partition(val topicPartition: TopicPartition,
    * leader's HW and LEO, the replica may become the leader before it fetches the committed data
    * and the data will be lost.
    *
-   * Technically, a replica shouldn't be in ISR if it hasn't caught up for longer than replicaLagTimeMaxMs,
-   * even if its log end offset is >= HW. However, to be consistent with how the follower determines
-   * whether a replica is in-sync, we only check HW.
+   * When the partition is under min ISR, the HW may be behind the leader's LEO. In that case, a
+   * replica is not considered caught up just because it has reached the HW. The same caught-up
+   * check used when shrinking the ISR is also required to avoid repeatedly expanding and shrinking
+   * the ISR for a follower that remains behind the leader's LEO.
    *
    * This function can be triggered when a replica's LEO has incremented.
    */
@@ -906,8 +907,13 @@ class Partition(val topicPartition: TopicPartition,
 
   private def isFollowerInSync(followerReplica: Replica): Boolean = {
     leaderLogIfLocal.exists { leaderLog =>
-      val followerEndOffset = followerReplica.stateSnapshot.logEndOffset
-      followerEndOffset >= leaderLog.highWatermark && leaderEpochStartOffsetOpt.exists(followerEndOffset >= _)
+      val followerState = followerReplica.stateSnapshot
+      val followerEndOffset = followerState.logEndOffset
+      val followerIsCaughtUp = !isUnderMinIsr ||
+        followerState.isCaughtUp(leaderLog.logEndOffset, time.milliseconds(), replicaLagTimeMaxMs)
+      followerEndOffset >= leaderLog.highWatermark &&
+        leaderEpochStartOffsetOpt.exists(followerEndOffset >= _) &&
+        followerIsCaughtUp
     }
   }
 

core/src/test/scala/unit/kafka/cluster/PartitionTest.scala

@@ -2283,6 +2283,63 @@ class PartitionTest extends AbstractPartitionTest {
     assertEquals(log.logEndOffset, partition.localLogOrException.highWatermark)
   }
 
+  @Test
+  def testIsrNotExpandedWhenFollowerHasNotCaughtUpToLeaderLogEnd(): Unit = {
+    configRepository.setTopicConfig(topicPartition.topic, TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "3")
+    val leaderEpoch = 5
+    val log = logManager.getOrCreateLog(topicPartition, topicId.toJava)
+
+    val remoteBrokerId = brokerId + 1
+    val replicas = Array(brokerId, remoteBrokerId, brokerId + 2)
+    val isr = Array(brokerId)
+    addBrokerEpochToMockMetadataCache(metadataCache, replicas)
+
+    val partition = new Partition(
+      topicPartition,
+      replicaLagTimeMaxMs = ReplicationConfigs.REPLICA_LAG_TIME_MAX_MS_DEFAULT,
+      localBrokerId = brokerId,
+      () => defaultBrokerEpoch(brokerId),
+      time,
+      alterPartitionListener,
+      delayedOperations,
+      metadataCache,
+      logManager,
+      alterPartitionManager
+    )
+    partition.createLogIfNotExists(isNew = false, isFutureReplica = false, offsetCheckpoints, topicId)
+
+    val partitionRegistration = new PartitionRegistration.Builder()
+      .setLeader(brokerId)
+      .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED)
+      .setLeaderEpoch(leaderEpoch)
+      .setIsr(isr)
+      .setPartitionEpoch(1)
+      .setReplicas(replicas)
+      .setDirectories(DirectoryId.unassignedArray(replicas.length))
+      .build()
+    assertTrue(partition.makeLeader(partitionRegistration, isNew = true, offsetCheckpoints, topicId),
+      "Expected become leader transition to succeed")
+    seedLogData(log, numRecords = 10, leaderEpoch = leaderEpoch)
+    time.sleep(partition.replicaLagTimeMaxMs + 1)
+
+    // The partition is under min ISR, so its HWM remains at zero. Reaching the HWM alone
+    // must not add a follower which is still behind the leader's log end offset.
+    fetchFollower(partition, replicaId = remoteBrokerId, fetchOffset = 1L)
+    assertReplicaState(partition, remoteBrokerId,
+      lastCaughtUpTimeMs = 0L,
+      logStartOffset = 0L,
+      logEndOffset = 1L
+    )
+    assertEquals(util.Set.of(brokerId), partition.partitionState.isr)
+    assertEquals(0, alterPartitionManager.isrUpdates.size)
+
+    // Once the follower catches up to the leader's log end, it can be added to the ISR.
+    fetchFollower(partition, replicaId = remoteBrokerId, fetchOffset = log.logEndOffset)
+    assertEquals(1, alterPartitionManager.isrUpdates.size)
+    assertEquals(util.Set.of[Integer](brokerId, remoteBrokerId),
+      alterPartitionManager.isrUpdates.peek().leaderAndIsr.isr)
+  }
+
   @Test
   def testAlterIsrLeaderAndIsrRace(): Unit = {
     val log = logManager.getOrCreateLog(topicPartition, topicId.toJava)