junrao commented on code in PR #14629:
URL: https://github.com/apache/kafka/pull/14629#discussion_r1375057023
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -53,25 +53,27 @@ object KafkaRequestHandler {
* 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
+ * @param requestLocal The RequestLocal for the current request handler
thread in case we need to call
+ * the callback function without queueing the callback
request
* @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) {
Review Comment:
> Hmmm. I'm not sure it makes sense to try to pass the request channel and
current in to every method we want to do a callback for.
Currently, the callback is only needed for produce request in
ReplicaManager. If you look at `KafkaApis.handleProduceRequest`, we already
pass in both request channel and current request to
`ReplicaManager.appendRecords` through `sendResponseCallback`.
> As for 2 and short circuiting... Are you suggesting I move the wrap call
into AddPartitionsToTxnManager? If so, I need to pass the appendEntries method
in there as well. That may be trickier with the typing.
Yes. Currently, we already pass in `appendEntries` as a callback to
`AddPartitionsToTxnManager`, right?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
