mumrah commented on a change in pull request #9732:
URL: https://github.com/apache/kafka/pull/9732#discussion_r545986703
##########
File path: core/src/main/scala/kafka/server/AlterIsrManager.scala
##########
@@ -34,25 +35,58 @@ import scala.collection.mutable
import scala.collection.mutable.ListBuffer
import scala.jdk.CollectionConverters._
+case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr, callback: Either[Errors, LeaderAndIsr] => Unit)
+
/**
* 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 AlterIsrManager {
- def start(): Unit
+ def start(): Unit = {}
+
+ def shutdown(): Unit = {}
def enqueue(alterIsrItem: AlterIsrItem): Boolean
def clearPending(topicPartition: TopicPartition): Unit
}
-case class AlterIsrItem(topicPartition: TopicPartition, leaderAndIsr:
LeaderAndIsr, callback: Either[Errors, LeaderAndIsr] => Unit)
+object AlterIsrManager {
+ def apply(
+ config: KafkaConfig,
+ metadataCache: MetadataCache,
+ scheduler: KafkaScheduler,
+ time: Time,
+ metrics: Metrics,
+ threadNamePrefix: Option[String],
+ brokerEpochSupplier: () => Long
+ ): AlterIsrManager = {
+ val channelManager = new BrokerToControllerChannelManager(
Review comment:
I guess we're not sharing these channel managers between anything. In
that case, moving them into the classes that need them seems fine.
##########
File path:
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala
##########
@@ -152,76 +151,89 @@ abstract class ControllerRequestCompletionHandler extends
RequestCompletionHandl
def onTimeout(): Unit
}
-case class BrokerToControllerQueueItem(request: AbstractRequest.Builder[_ <:
AbstractRequest],
- callback:
ControllerRequestCompletionHandler,
- deadlineMs: Long)
-
-class BrokerToControllerRequestThread(networkClient: KafkaClient,
- metadataUpdater: ManualMetadataUpdater,
- requestQueue:
LinkedBlockingDeque[BrokerToControllerQueueItem],
- metadataCache:
kafka.server.MetadataCache,
- config: KafkaConfig,
- listenerName: ListenerName,
- time: Time,
- threadName: String)
- extends InterBrokerSendThread(threadName, networkClient, time,
isInterruptible = false) {
-
+case class BrokerToControllerQueueItem(
+ createdTimeMs: Long,
+ request: AbstractRequest.Builder[_ <: AbstractRequest],
+ callback: ControllerRequestCompletionHandler
+)
+
+class BrokerToControllerRequestThread(
+ networkClient: KafkaClient,
+ metadataUpdater: ManualMetadataUpdater,
+ metadataCache: kafka.server.MetadataCache,
+ config: KafkaConfig,
+ listenerName: ListenerName,
+ time: Time,
+ threadName: String,
+ retryTimeoutMs: Long
+) extends InterBrokerSendThread(threadName, networkClient,
config.controllerSocketTimeoutMs, time, isInterruptible = false) {
+
+ private val requestQueue = new
LinkedBlockingDeque[BrokerToControllerQueueItem]()
private var activeController: Option[Node] = None
- override def requestTimeoutMs: Int = config.controllerSocketTimeoutMs
+ def enqueue(request: BrokerToControllerQueueItem): Unit = {
+ requestQueue.add(request)
+ if (activeController.isDefined) {
+ wakeup()
+ }
+ }
- override def generateRequests(): Iterable[RequestAndCompletionHandler] = {
- val requestsToSend = new mutable.Queue[RequestAndCompletionHandler]
- val topRequest = requestQueue.poll()
- if (topRequest != null) {
- val request = RequestAndCompletionHandler(
- activeController.get,
- topRequest.request,
- handleResponse(topRequest)
- )
+ def queueSize: Int = {
+ requestQueue.size
+ }
- requestsToSend.enqueue(request)
+ override def generateRequests(): Iterable[RequestAndCompletionHandler] = {
+ val currentTimeMs = time.milliseconds()
+ val requestIter = requestQueue.iterator()
Review comment:
Any reason to use Iterator here instead of queue methods (i.e., peek and
remove). Is it to ensure a consistent view of the queue while we're going
through it?
----------------------------------------------------------------
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
