junrao commented on code in PR #14629:
URL: https://github.com/apache/kafka/pull/14629#discussion_r1372156380
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -55,23 +55,23 @@ object KafkaRequestHandler {
* @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 = {
+ def wrap[T](fun: (RequestLocal, T) => Unit, requestLocal: RequestLocal): T
=> Unit = {
val requestChannel = threadRequestChannel.get()
val currentRequest = threadCurrentRequest.get()
if (requestChannel == null || currentRequest == null) {
if (!bypassThreadCheck)
throw new IllegalStateException("Attempted to reschedule to request
handler thread from non-request handler thread.")
- T => fun(T)
+ T => fun(requestLocal, T)
} else {
T => {
- if (threadCurrentRequest.get() != null) {
- // If the callback is actually executed on a request thread, we can
directly execute
+ if (threadCurrentRequest.get() == currentRequest) {
+ // If the callback is actually executed on the same request thread,
we can directly execute
Review Comment:
@jolshan : I wasn't suggesting simply skip the test
`threadCurrentRequest.get() == currentRequest`. I was wondering if we could
always add the callback to the `callbackQueue` through the following call,
independent of whether the caller is in the same request thread of not.
` requestChannel.sendCallbackRequest`
So, `wrap` will just be the following.
```
def wrap[T](fun: T => Unit): T => Unit = {
requestChannel.sendCallbackRequest(RequestChannel.CallbackRequest(() => fun(T),
currentRequest))
}
```
--
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
