hachikuji commented on a change in pull request #9732:
URL: https://github.com/apache/kafka/pull/9732#discussion_r541225075
##########
File path: core/src/main/scala/kafka/common/InterBrokerSendThread.scala
##########
@@ -32,17 +33,18 @@ import scala.jdk.CollectionConverters._
/**
* Class for inter-broker send thread that utilize a non-blocking network
client.
*/
-abstract class InterBrokerSendThread(name: String,
- networkClient: KafkaClient,
- time: Time,
- isInterruptible: Boolean = true)
- extends ShutdownableThread(name, isInterruptible) {
-
- def generateRequests(): Iterable[RequestAndCompletionHandler]
- def requestTimeoutMs: Int
+class InterBrokerSendThread(
+ name: String,
+ networkClient: KafkaClient,
+ requestTimeoutMs: Int,
+ time: Time,
+ isInterruptible: Boolean = true
+) extends ShutdownableThread(name, isInterruptible) {
+
+ private val inboundQueue = new
ConcurrentLinkedQueue[RequestAndCompletionHandler]()
Review comment:
The `BrokerToControllerChannelManager` is probably the only case where
this might be a concern since requests can be forwarded form the client. We
wouldn't want to block the request handler if the queue gets too large, but we
might want to start failing inbound requests. I'd suggest we create a separate
JIRA to think about this. Sound fair?
----------------------------------------------------------------
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
