hachikuji commented on a change in pull request #9564:
URL: https://github.com/apache/kafka/pull/9564#discussion_r527324620
##########
File path: core/src/main/scala/kafka/server/AlterIsrManager.scala
##########
@@ -88,20 +88,33 @@ class AlterIsrManagerImpl(val controllerChannelManager:
BrokerToControllerChanne
private def sendRequest(inflightAlterIsrItems: Seq[AlterIsrItem]): Unit = {
val message = buildRequest(inflightAlterIsrItems)
- def responseHandler(response: ClientResponse): Unit = {
- try {
- val body = response.responseBody().asInstanceOf[AlterIsrResponse]
- handleAlterIsrResponse(body, message.brokerEpoch(),
inflightAlterIsrItems)
- } finally {
- // Be sure to clear the in-flight flag to allow future AlterIsr
requests
- if (!inflightRequest.compareAndSet(true, false)) {
- throw new IllegalStateException("AlterIsr response callback called
when no requests were in flight")
+
+ def clearInflightRequests(): Unit = {
+ // Be sure to clear the in-flight flag to allow future AlterIsr requests
+ if (!inflightRequest.compareAndSet(true, false)) {
+ throw new IllegalStateException("AlterIsr response callback called
when no requests were in flight")
+ }
+ }
+
+ class AlterIsrResponseHandler extends
BrokerToControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ try {
+ val body = response.responseBody().asInstanceOf[AlterIsrResponse]
+ handleAlterIsrResponse(body, message.brokerEpoch(),
inflightAlterIsrItems)
+ } finally {
+ clearInflightRequests()
}
}
+
+ override def onFailure(exception: RuntimeException): Unit = {
+ error(s"Encountered $exception when sending AlterIsr to the
controller, clearing all pending states")
+ unsentIsrUpdates.clear()
Review comment:
There may be requests that never had a chance to be sent?
##########
File path:
core/src/main/scala/kafka/server/BrokerToControllerChannelManagerImpl.scala
##########
@@ -178,6 +194,10 @@ class BrokerToControllerRequestThread(networkClient:
KafkaClient,
}
}
+ private def isTimedOut(request: BrokerToControllerQueueItem): Boolean = {
+ (time.milliseconds() - request.firstEnqueuedTimeMs) > clientRequestTimeout
Review comment:
We do not want to timeout requests from the broker such as `AlterIsr`.
We only want this for client requests.
##########
File path: core/src/main/scala/kafka/server/AlterIsrManager.scala
##########
@@ -88,20 +88,33 @@ class AlterIsrManagerImpl(val controllerChannelManager:
BrokerToControllerChanne
private def sendRequest(inflightAlterIsrItems: Seq[AlterIsrItem]): Unit = {
val message = buildRequest(inflightAlterIsrItems)
- def responseHandler(response: ClientResponse): Unit = {
- try {
- val body = response.responseBody().asInstanceOf[AlterIsrResponse]
- handleAlterIsrResponse(body, message.brokerEpoch(),
inflightAlterIsrItems)
- } finally {
- // Be sure to clear the in-flight flag to allow future AlterIsr
requests
- if (!inflightRequest.compareAndSet(true, false)) {
- throw new IllegalStateException("AlterIsr response callback called
when no requests were in flight")
+
+ def clearInflightRequests(): Unit = {
+ // Be sure to clear the in-flight flag to allow future AlterIsr requests
+ if (!inflightRequest.compareAndSet(true, false)) {
+ throw new IllegalStateException("AlterIsr response callback called
when no requests were in flight")
+ }
+ }
+
+ class AlterIsrResponseHandler extends
BrokerToControllerRequestCompletionHandler {
+ override def onComplete(response: ClientResponse): Unit = {
+ try {
+ val body = response.responseBody().asInstanceOf[AlterIsrResponse]
+ handleAlterIsrResponse(body, message.brokerEpoch(),
inflightAlterIsrItems)
+ } finally {
+ clearInflightRequests()
}
}
+
+ override def onFailure(exception: RuntimeException): Unit = {
Review comment:
Don't we need to invoke the callbacks?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
