jolshan commented on code in PR #13391:
URL: https://github.com/apache/kafka/pull/13391#discussion_r1145376651
##########
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:
@artemlivshits thought this should be done. Perhaps he can explain better
than me.
--
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
