hachikuji commented on a change in pull request #9100:
URL: https://github.com/apache/kafka/pull/9100#discussion_r494480847
##########
File path: core/src/main/scala/kafka/controller/KafkaController.scala
##########
@@ -1764,6 +1769,145 @@ class KafkaController(val config: KafkaConfig,
}
}
+ def alterIsrs(alterIsrRequest: AlterIsrRequestData, callback:
AlterIsrResponseData => Unit): Unit = {
+ val isrsToAlter = mutable.Map[TopicPartition, LeaderAndIsr]()
+
+ alterIsrRequest.topics.forEach { topicReq =>
+ topicReq.partitions.forEach { partitionReq =>
+ val tp = new TopicPartition(topicReq.name, partitionReq.partitionIndex)
+ val newIsr = partitionReq.newIsr().asScala.toList.map(_.toInt)
+ isrsToAlter.put(tp, new LeaderAndIsr(alterIsrRequest.brokerId,
partitionReq.leaderEpoch, newIsr, partitionReq.currentIsrVersion))
+ }
+ }
+
+ def responseCallback(results: Either[Map[TopicPartition, Either[Errors,
LeaderAndIsr]], Errors]): Unit = {
+ val resp = new AlterIsrResponseData()
+ results match {
+ case Right(error) =>
+ resp.setErrorCode(error.code)
+ case Left(partitionResults) =>
+ resp.setTopics(new util.ArrayList())
+ partitionResults
+ .groupBy { case (tp, _) => tp.topic } // Group by topic
+ .foreach { case (topic, partitions) =>
+ // Add each topic part to the response
+ val topicResp = new AlterIsrResponseData.TopicData()
+ .setName(topic)
+ .setPartitions(new util.ArrayList())
+ resp.topics.add(topicResp)
+ partitions.foreach { case (tp, errorOrIsr) =>
+ // Add each partition part to the response (new ISR or error)
+ errorOrIsr match {
+ case Left(error) => topicResp.partitions.add(
+ new AlterIsrResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setErrorCode(error.code))
+ case Right(leaderAndIsr) => topicResp.partitions.add(
+ new AlterIsrResponseData.PartitionData()
+ .setPartitionIndex(tp.partition)
+ .setLeaderId(leaderAndIsr.leader)
+ .setLeaderEpoch(leaderAndIsr.leaderEpoch)
+ .setIsr(leaderAndIsr.isr.map(Integer.valueOf).asJava)
+ .setCurrentIsrVersion(leaderAndIsr.zkVersion))
+ }
+ }
+ }
+ }
+ callback.apply(resp)
+ }
+
+ eventManager.put(AlterIsrReceived(alterIsrRequest.brokerId,
alterIsrRequest.brokerEpoch, isrsToAlter, responseCallback))
+ }
+
+ private def processAlterIsr(brokerId: Int, brokerEpoch: Long, isrsToAlter:
Map[TopicPartition, LeaderAndIsr],
+ callback: AlterIsrCallback): Unit = {
+
+ // Handle a few short-circuits
+ if (!isActive) {
+ callback.apply(Right(Errors.NOT_CONTROLLER))
+ return
+ }
+
+ val brokerEpochOpt = controllerContext.liveBrokerIdAndEpochs.get(brokerId)
+ if (brokerEpochOpt.isEmpty) {
+ info(s"Ignoring AlterIsr due to unknown broker $brokerId")
+ callback.apply(Right(Errors.STALE_BROKER_EPOCH))
+ return
+ }
+
+ if (!brokerEpochOpt.contains(brokerEpoch)) {
+ info(s"Ignoring AlterIsr due to stale broker epoch $brokerEpoch for
broker $brokerId")
+ callback.apply(Right(Errors.STALE_BROKER_EPOCH))
+ return
+ }
+
+ val response = try {
+ val partitionResponses = mutable.HashMap[TopicPartition, Either[Errors,
LeaderAndIsr]]()
+
+ // Determine which partitions we will accept the new ISR for
+ val adjustedIsrs: Map[TopicPartition, LeaderAndIsr] =
isrsToAlter.flatMap {
+ case (tp: TopicPartition, newLeaderAndIsr: LeaderAndIsr) =>
+ val partitionError: Errors =
controllerContext.partitionLeadershipInfo(tp) match {
+ case Some(leaderIsrAndControllerEpoch) =>
+ val currentLeaderAndIsr =
leaderIsrAndControllerEpoch.leaderAndIsr
+ if (newLeaderAndIsr.leaderEpoch <
currentLeaderAndIsr.leaderEpoch) {
Review comment:
Consider the following scenario:
1) broker sends AlterIsr
2) the update succeeds but the response is lost
3) broker retries AlterIsr
Currently the leader will be stuck after 3) because it has no way to get the
latest LeaderAndIsr state if the first attempt fails. To handle this, I think
we need to add an idempotence check here. After we have validated the leader
epoch, if the intended state matches the current state, then we can just return
the current state.
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -1246,6 +1351,51 @@ class Partition(val topicPartition: TopicPartition,
}
}
+ private def sendAlterIsrRequest(proposedIsrState: IsrState): Boolean = {
+ val isrToSendOpt: Option[Set[Int]] = proposedIsrState match {
+ case PendingExpandIsr(isr, newInSyncReplicaId) => Some(isr +
newInSyncReplicaId)
+ case PendingShrinkIsr(isr, outOfSyncReplicaIds) => Some(isr --
outOfSyncReplicaIds)
+ case CommittedIsr(_) =>
+ error(s"Asked to send AlterIsr but there are no pending updates")
+ None
+ }
+ isrToSendOpt.exists { isrToSend =>
+ val newLeaderAndIsr = new LeaderAndIsr(localBrokerId, leaderEpoch,
isrToSend.toList, zkVersion)
+ val callbackPartial = handleAlterIsrResponse(isrToSend, _ :
Either[Errors, LeaderAndIsr])
+ alterIsrManager.enqueue(AlterIsrItem(topicPartition, newLeaderAndIsr,
callbackPartial))
+ }
+ }
+
+ private def handleAlterIsrResponse(proposedIsr: Set[Int], result:
Either[Errors, LeaderAndIsr]): Unit = {
+ inWriteLock(leaderIsrUpdateLock) {
+ result match {
+ case Left(error: Errors) => error match {
+ case Errors.UNKNOWN_TOPIC_OR_PARTITION =>
+ debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} since it doesn't know about this topic or
partition. Giving up.")
+ case Errors.FENCED_LEADER_EPOCH =>
+ debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} since we sent an old leader epoch. Giving up.")
+ case Errors.INVALID_UPDATE_VERSION =>
+ debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} due to invalid zk version. Retrying.")
+ sendAlterIsrRequest(isrState)
Review comment:
It might make more sense to handle this case similarly to
FENCED_LEADER_EPOCH. Retrying won't help since we know our version will be
rejected Come to think of it, this would be kind of a strange error to hit in
the current implementation which only allows one request inflight at a time.
For controller-initiated changes, we'd expect to hit FENCED_LEADER_EPOCH.
Anyway, I think it's still worth keeping the error.
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -1246,6 +1351,51 @@ class Partition(val topicPartition: TopicPartition,
}
}
+ private def sendAlterIsrRequest(proposedIsrState: IsrState): Boolean = {
+ val isrToSendOpt: Option[Set[Int]] = proposedIsrState match {
+ case PendingExpandIsr(isr, newInSyncReplicaId) => Some(isr +
newInSyncReplicaId)
+ case PendingShrinkIsr(isr, outOfSyncReplicaIds) => Some(isr --
outOfSyncReplicaIds)
+ case CommittedIsr(_) =>
+ error(s"Asked to send AlterIsr but there are no pending updates")
+ None
+ }
+ isrToSendOpt.exists { isrToSend =>
+ val newLeaderAndIsr = new LeaderAndIsr(localBrokerId, leaderEpoch,
isrToSend.toList, zkVersion)
+ val callbackPartial = handleAlterIsrResponse(isrToSend, _ :
Either[Errors, LeaderAndIsr])
+ alterIsrManager.enqueue(AlterIsrItem(topicPartition, newLeaderAndIsr,
callbackPartial))
+ }
+ }
+
+ private def handleAlterIsrResponse(proposedIsr: Set[Int], result:
Either[Errors, LeaderAndIsr]): Unit = {
+ inWriteLock(leaderIsrUpdateLock) {
+ result match {
+ case Left(error: Errors) => error match {
+ case Errors.UNKNOWN_TOPIC_OR_PARTITION =>
+ debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} since it doesn't know about this topic or
partition. Giving up.")
+ case Errors.FENCED_LEADER_EPOCH =>
+ debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} since we sent an old leader epoch. Giving up.")
+ case Errors.INVALID_UPDATE_VERSION =>
+ debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} due to invalid zk version. Retrying.")
+ sendAlterIsrRequest(isrState)
+ case _ =>
+ warn(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} due to $error. Retrying.")
+ sendAlterIsrRequest(isrState)
Review comment:
Is there any way that we could end up retrying after the pending ISR
state had already been reset? I know we have `AlterIsrManager.clearPending`,
but that only removes the partition from the unsent queue. How do we handle
inflight `AlterIsr` requests after the state has been reset. Seems like it
might be worth adding a check here to validate whether the request is still
needed.
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -1366,6 +1366,12 @@ class Partition(val topicPartition: TopicPartition,
}
}
+ /**
+ * This is called for each partition in the body of an AlterIsr response.
For errors which are non-retryable we simply
Review comment:
nit: conventionally we prefer "retriable"
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -1375,11 +1381,14 @@ class Partition(val topicPartition: TopicPartition,
case Errors.FENCED_LEADER_EPOCH =>
debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} since we sent an old leader epoch. Giving up.")
case Errors.INVALID_UPDATE_VERSION =>
- debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} due to invalid zk version. Retrying.")
- sendAlterIsrRequest(isrState)
+ debug(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} due to invalid zk version. Giving up.")
case _ =>
- warn(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} due to $error. Retrying.")
- sendAlterIsrRequest(isrState)
+ if (isrState.isInflight) {
+ warn(s"Controller failed to update ISR to
${proposedIsr.mkString(",")} due to $error. Retrying.")
+ sendAlterIsrRequest(isrState)
+ } else {
+ warn(s"Ignoring failed ISR update to
${proposedIsr.mkString(",")} since due to $error since we have a committed
ISR.")
Review comment:
nit (for follow-up): fix grammar "since due"
##########
File path: core/src/main/scala/kafka/controller/KafkaController.scala
##########
@@ -1847,21 +1847,22 @@ class KafkaController(val config: KafkaConfig,
// Determine which partitions we will accept the new ISR for
val adjustedIsrs: Map[TopicPartition, LeaderAndIsr] =
isrsToAlter.flatMap {
case (tp: TopicPartition, newLeaderAndIsr: LeaderAndIsr) =>
- val partitionError: Errors =
controllerContext.partitionLeadershipInfo(tp) match {
+ controllerContext.partitionLeadershipInfo(tp) match {
case Some(leaderIsrAndControllerEpoch) =>
val currentLeaderAndIsr =
leaderIsrAndControllerEpoch.leaderAndIsr
if (newLeaderAndIsr.leaderEpoch <
currentLeaderAndIsr.leaderEpoch) {
- Errors.FENCED_LEADER_EPOCH
+ partitionResponses(tp) = Left(Errors.FENCED_LEADER_EPOCH)
+ None
+ } else if (newLeaderAndIsr.equalsIgnoreZk(currentLeaderAndIsr)) {
+ // If a partition is already in the desired state, just return
it
Review comment:
It might be worth mentioning that this could happen in the case of a
retry after a successful update.
##########
File path: core/src/main/scala/kafka/server/KafkaApis.scala
##########
@@ -3066,20 +3066,18 @@ class KafkaApis(val requestChannel: RequestChannel,
def handleAlterIsrRequest(request: RequestChannel.Request): Unit = {
val alterIsrRequest = request.body[AlterIsrRequest]
- if (authorize(request.context, CLUSTER_ACTION, CLUSTER, CLUSTER_NAME)) {
- if (!controller.isActive) {
- sendResponseMaybeThrottle(request, requestThrottleMs =>
- alterIsrRequest.getErrorResponse(requestThrottleMs,
Errors.NOT_CONTROLLER.exception()))
- } else {
- controller.alterIsrs(alterIsrRequest.data,
- alterIsrResp => sendResponseMaybeThrottle(request, requestThrottleMs
=>
- new
AlterIsrResponse(alterIsrResp.setThrottleTimeMs(requestThrottleMs))
- )
- )
- }
- } else {
+ if (!authorize(request.context, CLUSTER_ACTION, CLUSTER, CLUSTER_NAME)) {
sendResponseMaybeThrottle(request, requestThrottleMs =>
alterIsrRequest.getErrorResponse(requestThrottleMs,
Errors.CLUSTER_AUTHORIZATION_FAILED.exception))
+ } else if (!controller.isActive) {
+ sendResponseMaybeThrottle(request, requestThrottleMs =>
+ alterIsrRequest.getErrorResponse(requestThrottleMs,
Errors.NOT_CONTROLLER.exception()))
Review comment:
nit: leave off parenthesis after `exception`
----------------------------------------------------------------
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
