hachikuji commented on a change in pull request #9912:
URL: https://github.com/apache/kafka/pull/9912#discussion_r561415958
##########
File path: core/src/main/scala/kafka/network/RequestChannel.scala
##########
@@ -371,9 +372,44 @@ class RequestChannel(val queueSize: Int,
requestQueue.put(request)
}
- /** Send a response back to the socket server to be sent over the network */
- def sendResponse(response: RequestChannel.Response): Unit = {
+ def closeConnection(
+ request: RequestChannel.Request,
+ errorCounts: java.util.Map[Errors, Integer]
+ ): Unit = {
+ // This case is used when the request handler has encountered an error,
but the client
+ // does not expect a response (e.g. when produce request has acks set to 0)
+ updateErrorMetrics(request.header.apiKey, errorCounts.asScala)
+ sendResponse(new RequestChannel.CloseConnectionResponse(request))
+ }
+
+ def sendResponse(
+ request: RequestChannel.Request,
+ response: AbstractResponse,
+ onComplete: Option[Send => Unit]
Review comment:
I decided not to do this here. I didn't like replacing `None` with `_ =>
{}` in uses, and neither did I like making the argument optional. The
alternative is to introduce a constant "no-op" function, but I found this also
a little awkward. If you think of a nice way to do it, I can review. I do think
it is better having a simpler type.
----------------------------------------------------------------
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
