andymg3 commented on code in PR #12499:
URL: https://github.com/apache/kafka/pull/12499#discussion_r941967046


##########
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##########
@@ -317,21 +373,24 @@ class BrokerToControllerRequestThread(
   override def generateRequests(): Iterable[RequestAndCompletionHandler] = {
     val currentTimeMs = time.milliseconds()
     val requestIter = requestQueue.iterator()
+    val controllerOpt = activeControllerOpt()
+
     while (requestIter.hasNext) {
       val request = requestIter.next
       if (currentTimeMs - request.createdTimeMs >= retryTimeoutMs) {
         requestIter.remove()
         request.callback.onTimeout()
       } else {
-        val controllerAddress = activeControllerAddress()
-        if (controllerAddress.isDefined) {
-          requestIter.remove()
-          return Some(RequestAndCompletionHandler(
-            time.milliseconds(),
-            controllerAddress.get,
-            request.request,
-            handleResponse(request)
-          ))
+        controllerOpt.foreach { activeController =>
+          if (activeController.epoch >= request.minControllerEpoch) {

Review Comment:
   To confirm, this check is done on the broker side right? I guess you sort of 
allude to this in the PR description that potentially a more ideal solution 
would be for the controller to do the check server side, but that would require 
a version bump. 



-- 
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

Reply via email to