[GitHub] [kafka] andymg3 commented on a diff in pull request #12499: KAFKA-14154; Ensure AlterPartition not sent to stale controller

[GitBox]

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


##
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##
@@ -364,15 +423,15 @@ class BrokerToControllerRequestThread(
   }
 
   override def doWork(): Unit = {
-if (activeControllerAddress().isDefined) {
+if (activeControllerOpt().isDefined) {
   super.pollOnce(Long.MaxValue)
 } else {
   debug("Controller isn't cached, looking for local metadata changes")
   controllerNodeProvider.get() match {
-case Some(controllerNode) =>
-  info(s"Recorded new controller, from now on will use broker 
$controllerNode")
-  updateControllerAddress(controllerNode)
-  metadataUpdater.setNodes(Seq(controllerNode).asJava)
+case Some(controllerNodeAndEpoch) =>

Review Comment:
   Is this where/how eventually the `LeaderAndIsr` from the new controller gets 
applied? 



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

[GitHub] [kafka] andymg3 commented on a diff in pull request #12499: KAFKA-14154; Ensure AlterPartition not sent to stale controller

[GitBox]

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


##
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##
@@ -364,15 +423,15 @@ class BrokerToControllerRequestThread(
   }
 
   override def doWork(): Unit = {
-if (activeControllerAddress().isDefined) {
+if (activeControllerOpt().isDefined) {
   super.pollOnce(Long.MaxValue)
 } else {
   debug("Controller isn't cached, looking for local metadata changes")
   controllerNodeProvider.get() match {
-case Some(controllerNode) =>
-  info(s"Recorded new controller, from now on will use broker 
$controllerNode")
-  updateControllerAddress(controllerNode)
-  metadataUpdater.setNodes(Seq(controllerNode).asJava)
+case Some(controllerNodeAndEpoch) =>

Review Comment:
   Is this where/how eventually the `LeaderAndIsr` from the new controller 
change gets applied? 



##
core/src/main/scala/kafka/server/BrokerToControllerChannelManager.scala:
##
@@ -364,15 +423,15 @@ class BrokerToControllerRequestThread(
   }
 
   override def doWork(): Unit = {
-if (activeControllerAddress().isDefined) {
+if (activeControllerOpt().isDefined) {
   super.pollOnce(Long.MaxValue)
 } else {
   debug("Controller isn't cached, looking for local metadata changes")
   controllerNodeProvider.get() match {
-case Some(controllerNode) =>
-  info(s"Recorded new controller, from now on will use broker 
$controllerNode")
-  updateControllerAddress(controllerNode)
-  metadataUpdater.setNodes(Seq(controllerNode).asJava)
+case Some(controllerNodeAndEpoch) =>

Review Comment:
   Is this where/how eventually the `LeaderAndIsr` from the new controllee gets 
applied? 



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

[GitHub] [kafka] andymg3 commented on a diff in pull request #12499: KAFKA-14154; Ensure AlterPartition not sent to stale controller

[GitBox]

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