hachikuji commented on a change in pull request #9100: URL: https://github.com/apache/kafka/pull/9100#discussion_r464705276
########## File path: core/src/main/scala/kafka/server/KafkaApis.scala ########## @@ -2956,6 +2956,22 @@ class KafkaApis(val requestChannel: RequestChannel, } } + def handleAlterIsrRequest(request: RequestChannel.Request): Unit = { + val alterIsrRequest = request.body[AlterIsrRequest] + + authorizeClusterOperation(request, CLUSTER_ACTION); + + // TODO do we need throttling for this response? Review comment: Probably reasonable to handle it the same way other inter-broker RPCs are handled. ########## File path: core/src/main/scala/kafka/controller/KafkaController.scala ########## @@ -1771,6 +1775,127 @@ class KafkaController(val config: KafkaConfig, } } + // TODO is it okay to pull message classes down into the controller? + def alterIsrs(alterIsrRequest: AlterIsrRequestData, callback: AlterIsrResponseData => Unit): Unit = { + //val brokerEpochOpt = controllerContext.liveBrokerIdAndEpochs.get(alterIsrRequest.brokerId()) + /*if (brokerEpochOpt.isEmpty) { + info(s"Ignoring AlterIsr due to unknown broker ${alterIsrRequest.brokerId()}") + // TODO is INVALID_REQUEST a reasonable error here? Review comment: Good question. Might be fair to assume the controller is correct and use STALE_BROKER_EPOCH. Once kip-500 is all done, it would be totally fair since the controller will be guaranteed to have the latest state. The other question is what the broker should do if it sees STALE_BROKER_EPOCH... ########## File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala ########## @@ -0,0 +1,121 @@ +package kafka.server + +import java.util +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicLong + +import kafka.api.LeaderAndIsr +import kafka.metrics.KafkaMetricsGroup +import kafka.utils.{Logging, Scheduler} +import kafka.zk.KafkaZkClient +import org.apache.kafka.clients.ClientResponse +import org.apache.kafka.common.TopicPartition +import org.apache.kafka.common.message.AlterIsrRequestData.{AlterIsrRequestPartitions, AlterIsrRequestTopics} +import org.apache.kafka.common.message.{AlterIsrRequestData, AlterIsrResponseData} +import org.apache.kafka.common.protocol.Errors +import org.apache.kafka.common.requests.{AlterIsrRequest, AlterIsrResponse} + +import scala.collection.mutable +import scala.jdk.CollectionConverters._ + +/** + * Handles the sending of AlterIsr requests to the controller. Updating the ISR is an asynchronous operation, + * so partitions will learn about updates through LeaderAndIsr messages sent from the controller + */ +trait AlterIsrChannelManager { + val IsrChangePropagationBlackOut = 5000L + val IsrChangePropagationInterval = 60000L + + def enqueueIsrUpdate(alterIsrItem: AlterIsrItem): Unit + + def startup(): Unit + + def shutdown(): Unit +} + +case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr: LeaderAndIsr, callback: Errors => Unit) + +class AlterIsrChannelManagerImpl(val controllerChannelManager: BrokerToControllerChannelManager, + val zkClient: KafkaZkClient, + val scheduler: Scheduler, + val brokerId: Int, + val brokerEpoch: Long) extends AlterIsrChannelManager with Logging with KafkaMetricsGroup { + + private val pendingIsrUpdates: mutable.Queue[AlterIsrItem] = new mutable.Queue[AlterIsrItem]() + private val lastIsrChangeMs = new AtomicLong(0) + private val lastIsrPropagationMs = new AtomicLong(0) + + override def enqueueIsrUpdate(alterIsrItem: AlterIsrItem): Unit = { + pendingIsrUpdates synchronized { + pendingIsrUpdates += alterIsrItem + lastIsrChangeMs.set(System.currentTimeMillis()) + } + } + + override def startup(): Unit = { + scheduler.schedule("alter-isr-send", maybePropagateIsrChanges _, period = 2500L, unit = TimeUnit.MILLISECONDS) Review comment: Hmm.. This adds a delay of 2.5s to every ISR change, which is a bit annoying. I guess the point is to allow batching? I think a better approach might be to send requests immediately on arrival, but set a limit on the maximum number of in-flight requests (maybe just 1) and let the changes accumulate when there is a request in-flight. Then we can still get a big batching benefit when there are a large number of ISR changes that need to be sent in a hurry. ########## File path: core/src/main/scala/kafka/controller/KafkaController.scala ########## @@ -1771,6 +1775,127 @@ class KafkaController(val config: KafkaConfig, } } + // TODO is it okay to pull message classes down into the controller? + def alterIsrs(alterIsrRequest: AlterIsrRequestData, callback: AlterIsrResponseData => Unit): Unit = { + //val brokerEpochOpt = controllerContext.liveBrokerIdAndEpochs.get(alterIsrRequest.brokerId()) + /*if (brokerEpochOpt.isEmpty) { + info(s"Ignoring AlterIsr due to unknown broker ${alterIsrRequest.brokerId()}") + // TODO is INVALID_REQUEST a reasonable error here? + callback.apply(new AlterIsrResponseData().setErrorCode(Errors.INVALID_REQUEST.code)) + return + } + + if (!brokerEpochOpt.contains(alterIsrRequest.brokerEpoch())) { + info(s"Ignoring AlterIsr due to stale broker epoch ${alterIsrRequest.brokerEpoch()} for broker ${alterIsrRequest.brokerId()}") + callback.apply(new AlterIsrResponseData().setErrorCode(Errors.STALE_BROKER_EPOCH.code)) + return + }*/ + + val isrsToAlter = mutable.Map[TopicPartition, LeaderAndIsr]() + + val resp = new AlterIsrResponseData() + resp.setTopics(new util.ArrayList()) + + alterIsrRequest.topics().forEach(topicReq => { + val topicResp = new AlterIsrResponseTopics() + .setName(topicReq.name()) + .setPartitions(new util.ArrayList()) + resp.topics().add(topicResp) + + topicReq.partitions().forEach(partitionReq => { + val partitionResp = new AlterIsrResponsePartitions() + .setPartitionIndex(partitionReq.partitionIndex()) + topicResp.partitions().add(partitionResp) + + // For each partition who's ISR we are altering, let's do some upfront validation for the broker response Review comment: Hmm.. Not sure it's worth doing these validations up-front. These checks could fail between the time that the event is enqueued and the time it is processed. ---------------------------------------------------------------- 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