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
