MINOR: Adjust `testClientDisconnectionUpdatesRequestMetrics` to also test small response case
Reviewers: Jun Rao Rajini Sivaram Andrew Choi
@@ -868,7 +868,16 @@ class SocketServerTest {
/* Test that we update request metrics if the client closes the connection while the broker response is in flight. */
@Test
- def testClientDisconnectionUpdatesRequestMetrics(): Unit = {
+ def testClientDisconnectionUpdatesRequestMetrics: Unit = {
+ // The way we detect a connection close from the client depends on the response size. If it's small, an
+ // IOException ("Connection reset by peer") is thrown when the Selector reads from the socket. If
+ // it's large, an IOException ("Broken pipe") is thrown when the Selector writes to the socket. We test
+ // both paths to ensure they are handled correctly.
+ checkClientDisconnectionUpdatesRequestMetrics(0)
+ checkClientDisconnectionUpdatesRequestMetrics(550000)
+ }
+
+ private def checkClientDisconnectionUpdatesRequestMetrics(responseBufferSize: Int): Unit = {
val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 0)
val serverMetrics = new Metrics
var conn: Socket = null
@@ -896,15 +905,10 @@ class SocketServerTest {
val requestMetrics = channel.metrics(request.header.apiKey.name)
def totalTimeHistCount(): Long = requestMetrics.totalTimeHist.count
- val expectedTotalTimeCount = totalTimeHistCount() + 1
+ val send = new NetworkSend(request.context.connectionId, ByteBuffer.allocate(responseBufferSize))
+ channel.sendResponse(new RequestChannel.SendResponse(request, send, Some("someResponse"), None))
- // send a large buffer to ensure that the broker detects the client disconnection while writing to the socket channel.
- // On Mac OS X, the initial write seems to always succeed and it is able to write up to 102400 bytes on the initial
- // write. If the buffer is smaller than this, the write is considered complete and the disconnection is not
- // detected. If the buffer is larger than 102400 bytes, a second write is attempted and it fails with an
- // IOException.
- val send = new NetworkSend(request.context.connectionId, ByteBuffer.allocate(550000))
- channel.sendResponse(new RequestChannel.SendResponse(request, send, None, None))
+ val expectedTotalTimeCount = totalTimeHistCount() + 1
TestUtils.waitUntilTrue(() => totalTimeHistCount() == expectedTotalTimeCount,
s"request metrics not updated, expected: $expectedTotalTimeCount, actual: ${totalTimeHistCount()}")
