hachikuji commented on a change in pull request #9100:
URL: https://github.com/apache/kafka/pull/9100#discussion_r469388095
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -748,7 +755,7 @@ class Partition(val topicPartition: TopicPartition,
leaderLogIfLocal match {
case Some(leaderLog) =>
// keep the current immutable replica list reference
- val curInSyncReplicaIds = inSyncReplicaIds
+ val curInSyncReplicaIds = effectiveInSyncReplicaIds
Review comment:
Related to the other comment, but we need to be careful with the min.isr
check below. I think it is correct to wait for `effectiveInSyncReplicaIds`
before acknowledging the produce request, but we should probably use the size
of `inSyncReplicaIds` in the min.isr check since that is the only set we can
guarantee.
##########
File path:
clients/src/main/java/org/apache/kafka/common/requests/AlterIsrRequest.java
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.message.AlterIsrRequestData;
+import org.apache.kafka.common.message.AlterIsrResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.protocol.types.Struct;
+
+public class AlterIsrRequest extends AbstractRequest {
+
+ private final AlterIsrRequestData data;
+
+ public AlterIsrRequest(AlterIsrRequestData data, short apiVersion) {
+ super(ApiKeys.ALTER_ISR, apiVersion);
+ this.data = data;
+ }
+
+ public AlterIsrRequestData data() {
+ return data;
+ }
+
+ @Override
+ protected Struct toStruct() {
+ return data.toStruct(version());
+ }
+
+ /**
+ * Get an error response for a request with specified throttle time in the
response if applicable
+ *
+ * @param throttleTimeMs
Review comment:
nit: maybe drop the parameters if they do not need to be documented
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -255,6 +255,10 @@ class Partition(val topicPartition: TopicPartition,
def isAddingReplica(replicaId: Int): Boolean =
assignmentState.isAddingReplica(replicaId)
+ // For advancing the HW we assume the largest ISR even if the controller
hasn't made the change yet
+ // This set includes the latest ISR (as we learned from LeaderAndIsr) and
any replicas from a pending ISR expansion
+ def effectiveInSyncReplicaIds: Set[Int] = inSyncReplicaIds |
pendingInSyncReplicaIds
Review comment:
We might need to be careful about performance here since this would get
called on every follower fetch.
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -485,13 +490,11 @@ class Partition(val topicPartition: TopicPartition,
def makeLeader(partitionState: LeaderAndIsrPartitionState,
Review comment:
There is a "classic" edge case in Kafka which goes as follows:
1. Leader is 1, ISR is [1, 2, 3]
2. Broker 3 begins controlled shutdown. While awaiting shutdown, it
continues fetching.
3. Controller bumps epoch and shrinks ISR to [1, 2] and notifies replicas
4. Before controlled shutdown completes and 3 stops fetching, the leader
adds it back to the ISR.
This bug was fixed by KIP-320 which added epoch validation to the Fetch API.
After shrinking the ISR in step 3, the controller will send `LeaderAndIsr` with
the updated epoch to [1, 2] and `StopReplica` to [3]. So 3 will not send any
fetches with the updated epoch, which means it's impossible for the leader to
add 3 back after observing the shrink to [1, 2].
I just want to make sure whether above is correct and whether `AlterIsr`
changes it in any way. I think the answer is no as long as ISR expansion is
_only_ done in response to a fetch request, but it's worth double-checking.
##########
File path: core/src/main/scala/kafka/controller/KafkaController.scala
##########
@@ -1756,6 +1761,141 @@ 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(partitionReq.leaderId(),
partitionReq.leaderEpoch(), newIsr, partitionReq.currentIsrVersion()))
+ }))
+
+ def responseCallback(results: Either[Map[TopicPartition, Errors],
Errors]): Unit = {
+ val resp = new AlterIsrResponseData()
+ results match {
+ case Right(error) =>
+ resp.setErrorCode(error.code())
Review comment:
nit: remove parenthesis for simpler getters like `code`. A few more of
these
##########
File path:
clients/src/main/java/org/apache/kafka/common/errors/InvalidIsrVersionException.java
##########
@@ -0,0 +1,29 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+public class InvalidIsrVersionException extends ApiException {
Review comment:
With KIP-500, I imagine we could end up with other cases where we end up
using optimistic concurrency control. Does it make sense to make this error a
little more generic? Maybe `INVALID_UPDATE_VERSION` or something like that..
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -990,6 +997,7 @@ class Partition(val topicPartition: TopicPartition,
leaderLogIfLocal match {
case Some(leaderLog) =>
val minIsr = leaderLog.config.minInSyncReplicas
+ // TODO is it safe to use pending ISR here?
Review comment:
I think the answer is no. The pending ISR set is not guaranteed, so we
cannot depend on it to enforce min.isr.
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -691,6 +700,7 @@ class Partition(val topicPartition: TopicPartition,
else
assignmentState = SimpleAssignmentState(assignment)
inSyncReplicaIds = isr
+ pendingInSyncReplicaIds = isr
Review comment:
The usage is a bit surprising given the "pending" name. I wonder if it
would be clearer if we used a type of `Option[Set[Int]]` so that we could use
`None` when there is no pending ISR change.
One more thing. It's worth double-checking the threading assumptions here.
It looks like `updateAssignmentAndIsr` is only called while holding the write
side of `leaderIsrUpdateLock`. On the other hand, I don't see any lock held in
`updateFollowerFetchState`. It's worth stepping through that logic to make sure
that we do not depend on `inSyncReplicaIds` and `pendingInSyncReplicaIds`
getting set atomically.
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -569,8 +575,8 @@ class Partition(val topicPartition: TopicPartition,
val oldLeaderEpoch = leaderEpoch
// record the epoch of the controller that made the leadership decision.
This is useful while updating the isr
// to maintain the decision maker controller's epoch in the zookeeper
path
- controllerEpoch = partitionState.controllerEpoch
+ info(s"Follower ignoring ISR for $topicPartition")
Review comment:
nit: info feels a bit high for a message like this
##########
File path: core/src/main/scala/kafka/cluster/Partition.scala
##########
@@ -807,7 +814,7 @@ class Partition(val topicPartition: TopicPartition,
var newHighWatermark = leaderLog.logEndOffsetMetadata
remoteReplicasMap.values.foreach { replica =>
if (replica.logEndOffsetMetadata.messageOffset <
newHighWatermark.messageOffset &&
- (curTime - replica.lastCaughtUpTimeMs <= replicaLagTimeMaxMs ||
inSyncReplicaIds.contains(replica.brokerId))) {
+ (curTime - replica.lastCaughtUpTimeMs <= replicaLagTimeMaxMs ||
effectiveInSyncReplicaIds.contains(replica.brokerId))) {
Review comment:
I think it's worth adding a comment in the cases we rely on
`effectiveInSyncReplicaIds` to explain why.
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
+import java.util.concurrent.{ScheduledFuture, 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
Review comment:
nit: avoid loaded terminology like "blackout" (see
https://cwiki.apache.org/confluence/display/KAFKA/KIP-629%3A+Use+racially+neutral+terms+in+our+codebase).
Do we actually need this or `IsrChangePropagationInterval` below?
##########
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 think the implementation here is actually different than what was in
the model. Consider the following case:
1) Initial state: isr=[1, 2], pendingIsr=[1, 2]
2) Leader expands ISR. isr=[1, 2], pendingIsr=[1, 2, 3]
3) Leader shrinks ISR. isr=[1, 2], pendingIsr=[1, 2]
We don't know which of the updates in 2) or 3) will be accepted, but after
3), we will not assume that broker 3 could be in the ISR, which could lead to a
correctness violation if the update in 2) is accepted by the controller.
In the model, we always assumed the maximal ISR across _any_ potential
update to protect from this edge case. Maybe in the end it is simpler to not
allow multiple in-flight updates.
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
+import java.util.concurrent.{ScheduledFuture, 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 clearPending(topicPartition: TopicPartition): Unit
+
+ def startup(): Unit
+
+ def shutdown(): Unit
+}
+
+case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr)
+
+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.Map[TopicPartition, AlterIsrItem] =
new mutable.HashMap[TopicPartition, AlterIsrItem]()
+ private val lastIsrChangeMs = new AtomicLong(0)
+ private val lastIsrPropagationMs = new AtomicLong(0)
+
+ @volatile private var scheduledRequest: Option[ScheduledFuture[_]] = None
+
+ override def enqueueIsrUpdate(alterIsrItem: AlterIsrItem): Unit = {
+ pendingIsrUpdates synchronized {
+ pendingIsrUpdates(alterIsrItem.topicPartition) = alterIsrItem
+ lastIsrChangeMs.set(System.currentTimeMillis())
Review comment:
We should use `Time`
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
+import java.util.concurrent.{ScheduledFuture, 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 clearPending(topicPartition: TopicPartition): Unit
+
+ def startup(): Unit
+
+ def shutdown(): Unit
+}
+
+case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr)
+
+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.Map[TopicPartition, AlterIsrItem] =
new mutable.HashMap[TopicPartition, AlterIsrItem]()
+ private val lastIsrChangeMs = new AtomicLong(0)
+ private val lastIsrPropagationMs = new AtomicLong(0)
+
+ @volatile private var scheduledRequest: Option[ScheduledFuture[_]] = None
+
+ override def enqueueIsrUpdate(alterIsrItem: AlterIsrItem): Unit = {
+ pendingIsrUpdates synchronized {
+ pendingIsrUpdates(alterIsrItem.topicPartition) = alterIsrItem
+ lastIsrChangeMs.set(System.currentTimeMillis())
+ // Rather than sending right away, we'll delay at most 50ms to allow for
batching of ISR changes happening
+ // in fast succession
+ if (scheduledRequest.isEmpty) {
+ scheduledRequest = Some(scheduler.schedule("propagate-alter-isr",
propagateIsrChanges, 50, -1, TimeUnit.MILLISECONDS))
+ }
+ }
+ }
+
+ override def clearPending(topicPartition: TopicPartition): Unit = {
+ pendingIsrUpdates synchronized {
+ // when we get a new LeaderAndIsr, we clear out any pending requests
+ pendingIsrUpdates.remove(topicPartition)
+ }
+ }
+
+ override def startup(): Unit = {
+ controllerChannelManager.start()
+ }
+
+ override def shutdown(): Unit = {
+ controllerChannelManager.shutdown()
+ }
+
+ private def propagateIsrChanges(): Unit = {
+ val now = System.currentTimeMillis()
+ pendingIsrUpdates synchronized {
+ if (pendingIsrUpdates.nonEmpty) {
+ // Max ISRs to send?
+ val message = new AlterIsrRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch)
+ .setTopics(new util.ArrayList())
+
+
pendingIsrUpdates.values.groupBy(_.topicPartition.topic()).foreachEntry((topic,
items) => {
+ val topicPart = new AlterIsrRequestTopics()
+ .setName(topic)
+ .setPartitions(new util.ArrayList())
+ message.topics().add(topicPart)
+ items.foreach(item => {
+ topicPart.partitions().add(new AlterIsrRequestPartitions()
+ .setPartitionIndex(item.topicPartition.partition())
+ .setLeaderId(item.leaderAndIsr.leader)
+ .setLeaderEpoch(item.leaderAndIsr.leaderEpoch)
+ .setNewIsr(item.leaderAndIsr.isr.map(Integer.valueOf).asJava)
+ .setCurrentIsrVersion(item.leaderAndIsr.zkVersion)
+ )
+ })
+ })
+
+ def responseHandler(response: ClientResponse): Unit = {
Review comment:
I think the basic approach here is to ignore successful responses and
wait for the `LeaderAndIsr` update. I am wondering how we should handle the
case when the update failed. Say for example that our update fails with the
INVALID_VERSION error. Inside `Partition`, we will still have the pendingIsr
set. Do we need to clear it? How about other errors?
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
+import java.util.concurrent.{ScheduledFuture, 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 clearPending(topicPartition: TopicPartition): Unit
+
+ def startup(): Unit
+
+ def shutdown(): Unit
+}
+
+case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr)
+
+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.Map[TopicPartition, AlterIsrItem] =
new mutable.HashMap[TopicPartition, AlterIsrItem]()
+ private val lastIsrChangeMs = new AtomicLong(0)
+ private val lastIsrPropagationMs = new AtomicLong(0)
+
+ @volatile private var scheduledRequest: Option[ScheduledFuture[_]] = None
+
+ override def enqueueIsrUpdate(alterIsrItem: AlterIsrItem): Unit = {
+ pendingIsrUpdates synchronized {
+ pendingIsrUpdates(alterIsrItem.topicPartition) = alterIsrItem
+ lastIsrChangeMs.set(System.currentTimeMillis())
+ // Rather than sending right away, we'll delay at most 50ms to allow for
batching of ISR changes happening
+ // in fast succession
+ if (scheduledRequest.isEmpty) {
+ scheduledRequest = Some(scheduler.schedule("propagate-alter-isr",
propagateIsrChanges, 50, -1, TimeUnit.MILLISECONDS))
+ }
+ }
+ }
+
+ override def clearPending(topicPartition: TopicPartition): Unit = {
+ pendingIsrUpdates synchronized {
+ // when we get a new LeaderAndIsr, we clear out any pending requests
+ pendingIsrUpdates.remove(topicPartition)
Review comment:
Removal from this set won't prevent `BrokerToControllerRequestThread`
from retrying in-flight requests. I'm considering whether we should have a way
to cancel requests that we are still awaiting.
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
+import java.util.concurrent.{ScheduledFuture, 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 clearPending(topicPartition: TopicPartition): Unit
+
+ def startup(): Unit
+
+ def shutdown(): Unit
+}
+
+case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr)
+
+class AlterIsrChannelManagerImpl(val controllerChannelManager:
BrokerToControllerChannelManager,
+ val zkClient: KafkaZkClient,
+ val scheduler: Scheduler,
+ val brokerId: Int,
+ val brokerEpoch: Long) extends
AlterIsrChannelManager with Logging with KafkaMetricsGroup {
Review comment:
The broker epoch is not a constant. It gets reinitialized whenever the
broker has to create a new session.
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
Review comment:
Missing license header in this file.
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
+import java.util.concurrent.{ScheduledFuture, 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 clearPending(topicPartition: TopicPartition): Unit
+
+ def startup(): Unit
+
+ def shutdown(): Unit
+}
+
+case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr)
+
+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.Map[TopicPartition, AlterIsrItem] =
new mutable.HashMap[TopicPartition, AlterIsrItem]()
+ private val lastIsrChangeMs = new AtomicLong(0)
+ private val lastIsrPropagationMs = new AtomicLong(0)
+
+ @volatile private var scheduledRequest: Option[ScheduledFuture[_]] = None
+
+ override def enqueueIsrUpdate(alterIsrItem: AlterIsrItem): Unit = {
+ pendingIsrUpdates synchronized {
+ pendingIsrUpdates(alterIsrItem.topicPartition) = alterIsrItem
+ lastIsrChangeMs.set(System.currentTimeMillis())
+ // Rather than sending right away, we'll delay at most 50ms to allow for
batching of ISR changes happening
+ // in fast succession
+ if (scheduledRequest.isEmpty) {
+ scheduledRequest = Some(scheduler.schedule("propagate-alter-isr",
propagateIsrChanges, 50, -1, TimeUnit.MILLISECONDS))
+ }
+ }
+ }
+
+ override def clearPending(topicPartition: TopicPartition): Unit = {
+ pendingIsrUpdates synchronized {
+ // when we get a new LeaderAndIsr, we clear out any pending requests
+ pendingIsrUpdates.remove(topicPartition)
+ }
+ }
+
+ override def startup(): Unit = {
+ controllerChannelManager.start()
+ }
+
+ override def shutdown(): Unit = {
+ controllerChannelManager.shutdown()
+ }
+
+ private def propagateIsrChanges(): Unit = {
+ val now = System.currentTimeMillis()
+ pendingIsrUpdates synchronized {
+ if (pendingIsrUpdates.nonEmpty) {
+ // Max ISRs to send?
+ val message = new AlterIsrRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch)
+ .setTopics(new util.ArrayList())
+
+
pendingIsrUpdates.values.groupBy(_.topicPartition.topic()).foreachEntry((topic,
items) => {
+ val topicPart = new AlterIsrRequestTopics()
+ .setName(topic)
+ .setPartitions(new util.ArrayList())
+ message.topics().add(topicPart)
+ items.foreach(item => {
+ topicPart.partitions().add(new AlterIsrRequestPartitions()
+ .setPartitionIndex(item.topicPartition.partition())
+ .setLeaderId(item.leaderAndIsr.leader)
+ .setLeaderEpoch(item.leaderAndIsr.leaderEpoch)
+ .setNewIsr(item.leaderAndIsr.isr.map(Integer.valueOf).asJava)
+ .setCurrentIsrVersion(item.leaderAndIsr.zkVersion)
+ )
+ })
+ })
+
+ def responseHandler(response: ClientResponse): Unit = {
+
println(response.responseBody().toString(response.requestHeader().apiVersion()))
+ val body: AlterIsrResponse =
response.responseBody().asInstanceOf[AlterIsrResponse]
+ val data: AlterIsrResponseData = body.data()
+ Errors.forCode(data.errorCode()) match {
+ case Errors.NONE => info(s"Controller handled AlterIsr request")
+ case e: Errors => warn(s"Controller returned an error when
handling AlterIsr request: $e")
+ }
+ data.topics().forEach(topic => {
+ topic.partitions().forEach(topicPartition => {
+ Errors.forCode(topicPartition.errorCode()) match {
+ case Errors.NONE => info(s"Controller handled AlterIsr for
$topicPartition")
+ case e: Errors => warn(s"Controlled had an error handling
AlterIsr for $topicPartition: $e")
+ }
+ })
+ })
+ }
+
+ info("Sending AlterIsr to controller")
Review comment:
nit: more useful as a debug if we add request details to the message
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
+import java.util.concurrent.{ScheduledFuture, 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 clearPending(topicPartition: TopicPartition): Unit
+
+ def startup(): Unit
+
+ def shutdown(): Unit
+}
+
+case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr)
+
+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.Map[TopicPartition, AlterIsrItem] =
new mutable.HashMap[TopicPartition, AlterIsrItem]()
+ private val lastIsrChangeMs = new AtomicLong(0)
+ private val lastIsrPropagationMs = new AtomicLong(0)
+
+ @volatile private var scheduledRequest: Option[ScheduledFuture[_]] = None
+
+ override def enqueueIsrUpdate(alterIsrItem: AlterIsrItem): Unit = {
+ pendingIsrUpdates synchronized {
+ pendingIsrUpdates(alterIsrItem.topicPartition) = alterIsrItem
+ lastIsrChangeMs.set(System.currentTimeMillis())
+ // Rather than sending right away, we'll delay at most 50ms to allow for
batching of ISR changes happening
+ // in fast succession
+ if (scheduledRequest.isEmpty) {
+ scheduledRequest = Some(scheduler.schedule("propagate-alter-isr",
propagateIsrChanges, 50, -1, TimeUnit.MILLISECONDS))
+ }
+ }
+ }
+
+ override def clearPending(topicPartition: TopicPartition): Unit = {
+ pendingIsrUpdates synchronized {
+ // when we get a new LeaderAndIsr, we clear out any pending requests
+ pendingIsrUpdates.remove(topicPartition)
+ }
+ }
+
+ override def startup(): Unit = {
+ controllerChannelManager.start()
+ }
+
+ override def shutdown(): Unit = {
+ controllerChannelManager.shutdown()
+ }
+
+ private def propagateIsrChanges(): Unit = {
+ val now = System.currentTimeMillis()
+ pendingIsrUpdates synchronized {
+ if (pendingIsrUpdates.nonEmpty) {
+ // Max ISRs to send?
+ val message = new AlterIsrRequestData()
+ .setBrokerId(brokerId)
+ .setBrokerEpoch(brokerEpoch)
+ .setTopics(new util.ArrayList())
+
+
pendingIsrUpdates.values.groupBy(_.topicPartition.topic()).foreachEntry((topic,
items) => {
+ val topicPart = new AlterIsrRequestTopics()
+ .setName(topic)
+ .setPartitions(new util.ArrayList())
+ message.topics().add(topicPart)
+ items.foreach(item => {
+ topicPart.partitions().add(new AlterIsrRequestPartitions()
+ .setPartitionIndex(item.topicPartition.partition())
+ .setLeaderId(item.leaderAndIsr.leader)
+ .setLeaderEpoch(item.leaderAndIsr.leaderEpoch)
+ .setNewIsr(item.leaderAndIsr.isr.map(Integer.valueOf).asJava)
+ .setCurrentIsrVersion(item.leaderAndIsr.zkVersion)
+ )
+ })
+ })
+
+ def responseHandler(response: ClientResponse): Unit = {
+
println(response.responseBody().toString(response.requestHeader().apiVersion()))
+ val body: AlterIsrResponse =
response.responseBody().asInstanceOf[AlterIsrResponse]
+ val data: AlterIsrResponseData = body.data()
+ Errors.forCode(data.errorCode()) match {
+ case Errors.NONE => info(s"Controller handled AlterIsr request")
Review comment:
Probably need to reduce the log level here and below.
##########
File path: core/src/main/scala/kafka/server/AlterIsrChannelManager.scala
##########
@@ -0,0 +1,132 @@
+package kafka.server
+
+import java.util
+import java.util.concurrent.{ScheduledFuture, 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 clearPending(topicPartition: TopicPartition): Unit
+
+ def startup(): Unit
+
+ def shutdown(): Unit
+}
+
+case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr)
+
+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.Map[TopicPartition, AlterIsrItem] =
new mutable.HashMap[TopicPartition, AlterIsrItem]()
Review comment:
The term "pending" again is a little unclear. Perhaps "unsentIsrUpdates"
would make the usage clearer.
----------------------------------------------------------------
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
