junrao commented on code in PR #13391: URL: https://github.com/apache/kafka/pull/13391#discussion_r1163446164
########## core/src/main/scala/kafka/server/KafkaRequestHandler.scala: ########## @@ -69,20 +108,52 @@ class KafkaRequestHandler(id: Int, completeShutdown() return + case callback: RequestChannel.CallbackRequest => + try { + val originalRequest = callback.originalRequest + + // If we've already executed a callback for this request, reset the times and subtract the callback time from the + // new dequeue time. This will allow calculation of multiple callback times. + // Otherwise, set dequeue time to now. + if (originalRequest.callbackRequestDequeueTimeNanos.isDefined) { + val prevCallbacksTimeNanos = originalRequest.callbackRequestCompleteTimeNanos.getOrElse(0L) - originalRequest.callbackRequestDequeueTimeNanos.getOrElse(0L) + originalRequest.callbackRequestCompleteTimeNanos = None + originalRequest.callbackRequestDequeueTimeNanos = Some(time.nanoseconds() - prevCallbacksTimeNanos) + } else { + originalRequest.callbackRequestDequeueTimeNanos = Some(time.nanoseconds()) + } + + currentRequest.set(originalRequest) + callback.fun() + if (originalRequest.callbackRequestCompleteTimeNanos.isEmpty) Review Comment: I am still a bit confused on this. case a - only 1 callback. In this case, callbackRequestCompleteTimeNanos is never set and is expected to be empty. case b - multiple callbacks. In this case, after the first callback, callbackRequestDequeueTimeNanos is set. So, when the 2nd callback is processed, we set callbackRequestCompleteTimeNanos to empty before executing the callback. When we get here, callbackRequestCompleteTimeNanos is expected to be empty. So, it seems that in both case a and b, we expect callbackRequestCompleteTimeNanos to be empty? -- 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