junrao commented on code in PR #13391:
URL: https://github.com/apache/kafka/pull/13391#discussion_r1146944915
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -35,6 +36,40 @@ trait ApiRequestHandler {
def handle(request: RequestChannel.Request, requestLocal: RequestLocal): Unit
}
+object KafkaRequestHandler {
+ // Support for scheduling callbacks on a request thread.
+ // TODO: we may want to pass more request context, e.g. processor id (see
RequestChannel.Request)
+ private val threadRequestChannel = new ThreadLocal[RequestChannel]
+
+ // For testing
+ private var bypassThreadCheck = false
+ def setBypassThreadCheck(bypassCheck: Boolean): Unit = {
+ bypassThreadCheck = bypassCheck
+ }
+
+ /**
+ * Wrap callback to schedule it on a request thread.
+ * NOTE: this function must be called on a request thread.
+ * @param fun Callback function to execute
+ * @return Wrapped callback that would execute `fun` on a request thread
+ */
+ def wrap[T](fun: T => Unit): T => Unit = {
+ val requestChannel = threadRequestChannel.get()
+ if (requestChannel == null) {
+ if (!bypassThreadCheck)
+ throw new IllegalStateException("Attempted to reschedule to request
handler thread from non-request handler thread.")
+ T => fun(T)
+ } else {
+ T => {
+ // The requestChannel is captured in this lambda, so when it's
executed on the callback thread
+ // we can re-schedule the original callback on a request thread.
Review Comment:
Chatted with Artem offline. His reasoning is for performance. It's better to
do any IO related operations in the request thread pool to prevent blocking the
callback thread. This could be a bit better. If we do this, maybe we should
also change TransactionMarkerRequestCompletionHandler so that it writes the
complete marker in the request thread instead of the callback thread. That
could be done in a followup jira.
--
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
