artemlivshits commented on code in PR #14629:
URL: https://github.com/apache/kafka/pull/14629#discussion_r1372379141
##########
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:
My understanding is that concurrency is not a problem in production logic
(i.e. the fact that callback might actually execute at any time before or after
or concurrently with the passing thread), but unit tests rely on deterministic
order of execution and so adding concurrency here makes them "flaky".
I agree in principle that if we could fold this logic into just one case
(i.e. always schedule on a new request thread), it would make it simpler; when
this change was proposed I looked into unit test fix and seemed more involved
than adding this logic.
--
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]
