mumrah commented on a change in pull request #9100: URL: https://github.com/apache/kafka/pull/9100#discussion_r469467881
########## File path: core/src/main/scala/kafka/cluster/Partition.scala ########## @@ -1210,28 +1218,20 @@ class Partition(val topicPartition: TopicPartition, } } - private def expandIsr(newIsr: Set[Int]): Unit = { - val newLeaderAndIsr = new LeaderAndIsr(localBrokerId, leaderEpoch, newIsr.toList, zkVersion) - val zkVersionOpt = stateStore.expandIsr(controllerEpoch, newLeaderAndIsr) - maybeUpdateIsrAndVersion(newIsr, zkVersionOpt) - } + private def expandIsr(newInSyncReplica: Int): Unit = { + pendingInSyncReplicaIds += newInSyncReplica + info(s"Adding new in-sync replica $newInSyncReplica. Pending ISR updated to [${pendingInSyncReplicaIds.mkString(",")}]") - private[cluster] def shrinkIsr(newIsr: Set[Int]): Unit = { - val newLeaderAndIsr = new LeaderAndIsr(localBrokerId, leaderEpoch, newIsr.toList, zkVersion) - val zkVersionOpt = stateStore.shrinkIsr(controllerEpoch, newLeaderAndIsr) - maybeUpdateIsrAndVersion(newIsr, zkVersionOpt) + val newLeaderAndIsr = new LeaderAndIsr(localBrokerId, leaderEpoch, pendingInSyncReplicaIds.toList, zkVersion) + alterIsrChannelManager.enqueueIsrUpdate(AlterIsrItem(topicPartition, newLeaderAndIsr)) } - private[cluster] def maybeUpdateIsrAndVersion(isr: Set[Int], zkVersionOpt: Option[Int]): Unit = { - zkVersionOpt match { - case Some(newVersion) => - inSyncReplicaIds = isr - zkVersion = newVersion - info("ISR updated to [%s] and zkVersion updated to [%d]".format(isr.mkString(","), zkVersion)) + private[cluster] def shrinkIsr(outOfSyncReplicas: Set[Int]): Unit = { + pendingInSyncReplicaIds --= outOfSyncReplicas Review comment: I'm currently looking at the effective ISR to find new out of sync replicas. This can include new ISR members which haven't made it into the "true" ISR via LeaderAndIsr yet (like broker=3 in your example). Maybe we should only consider removing ISR members iff they are in the true ISR. IOW changing from ```scala val candidateReplicaIds = effectiveInSyncReplicaIds - localBrokerId ``` to ```scala val candidateReplicaIds = inSyncReplicaIds - localBrokerId ``` Also, I wonder if the batching that's happening in AlterIsrChannelManager violates the model. It sends the request asynchronously with a small delay, so multiple ISR changes can be batched into one AlterIsr. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org