artemlivshits commented on code in PR #13391:
URL: https://github.com/apache/kafka/pull/13391#discussion_r1160975587
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -110,8 +110,23 @@ class KafkaRequestHandler(id: Int,
case callback: RequestChannel.CallbackRequest =>
try {
- callback.originalRequest.callbackRequestDequeTimeNanos =
Some(time.nanoseconds())
+ 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)
Review Comment:
Need to clear as well in the finally clause.
##########
core/src/main/scala/kafka/server/KafkaRequestHandler.scala:
##########
@@ -110,8 +110,23 @@ class KafkaRequestHandler(id: Int,
case callback: RequestChannel.CallbackRequest =>
try {
- callback.originalRequest.callbackRequestDequeTimeNanos =
Some(time.nanoseconds())
+ 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) {
Review Comment:
Not sure if the 'if' clause is required, prevCallbacksTimeNanos would be 0
if there was no prev callback.
--
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
