artemlivshits commented on code in PR #13463:
URL: https://github.com/apache/kafka/pull/13463#discussion_r1152665868
##########
core/src/main/scala/kafka/common/InterBrokerSendThread.scala:
##########
@@ -28,12 +26,14 @@ import org.apache.kafka.common.requests.AbstractRequest
import org.apache.kafka.common.utils.Time
import org.apache.kafka.server.util.ShutdownableThread
+import java.util.Map.Entry
+import java.util.{ArrayDeque, ArrayList, Collection, Collections, HashMap,
Iterator}
import scala.jdk.CollectionConverters._
/**
* Class for inter-broker send thread that utilize a non-blocking network
client.
*/
-abstract class InterBrokerSendThread(
+class InterBrokerSendThread(
Review Comment:
Maybe we should name it InterBrokerSender? The fact that it has only one
thread is more like an internal implementation detail.
##########
core/src/main/scala/kafka/common/InterBrokerSendThread.scala:
##########
@@ -156,6 +162,13 @@ abstract class InterBrokerSendThread(
def wakeup(): Unit = networkClient.wakeup()
}
+abstract class InterBrokerRequestManager(interBrokerSendThread:
InterBrokerSendThread) {
Review Comment:
I wonder if we could just set the interBrokerSendThread in the
.addRequestManager method. This way, we don't need to pass
interBrokerSendThread in all request manager classes (and wakeup doesn't make
sense anyway until the request manager is added to the interBrokerSendThread).
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
