KAFKA-19241: Updated tests in ShareFetchAcknowledgeRequestTest to reuse the socket for subsequent requests #19640
Conversation
…se the same socket in subsequent requests
github-actions
bot
added
triage
AndrewJSchofield
added
KIP-932
| @@ -913,6 +902,20 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) { | |||
| assertEquals(expectedResponseData.results.asScala.toSet, deleteGroupsResponse.data.results.asScala.toSet) | |||
| } | |||
|
|
|||
| protected def createSocket(): Socket = { | |||
There was a problem hiding this comment.
Shouldn't it be connectAny?
| ) | ||
| } | ||
|
|
||
| protected def createSocket(destination: Int): Socket = { |
There was a problem hiding this comment.
Shouldn't it be connect?
| protected def sendAndReceiveFromExistingSocket[T <: AbstractResponse]( | ||
| request: AbstractRequest, | ||
| destination: Int | ||
| )(implicit classTag: ClassTag[T]): T = { | ||
| val socket = IntegrationTestUtils.connect(brokerSocketServer(destination), cluster.clientListener()) | ||
| openSockets += socket | ||
| IntegrationTestUtils.sendAndReceive[T](request, socket) | ||
| } | ||
|
|
||
| protected def connectAndReceiveWithoutClosingSocket[T <: AbstractResponse]( | ||
| request: AbstractRequest | ||
| socket: Socket | ||
| )(implicit classTag: ClassTag[T]): T = { | ||
| val socket = IntegrationTestUtils.connect(cluster.anyBrokerSocketServer(), cluster.clientListener()) | ||
| openSockets += socket | ||
| IntegrationTestUtils.sendAndReceive[T](request, socket) | ||
| } |
There was a problem hiding this comment.
Why do you need this method and can't use IntegrationTestUtils.sendAndReceive[T](request, socket) directly?
|
|
||
| assertEquals(Errors.UNSUPPORTED_VERSION.code, shareFetchResponse.data.errorCode) | ||
| assertEquals(0, shareFetchResponse.data.acquisitionLockTimeoutMs) | ||
|
|
||
| socket.close() |
There was a problem hiding this comment.
Is there a way we can have a check in tearDown that all open sockets are closed from the tests?
There was a problem hiding this comment.
Thanks for the review. I have now updated the tests to include a ListBuffer that stores any new sockets that are created. And in the tearDown, the list is iterated to close all the sockets that are still open
There was a problem hiding this comment.
So if we avoid this call here and rely on tearDown to close sockets then will it cause any trouble? Then you don't need to have an explicit if (!socket.isClosed) as well?
| if (!socket.isClosed) { | ||
| socket.close() | ||
| } |
There was a problem hiding this comment.
Seems strange that we are fetching from openSockets and checking if it's closed. So if it was closed then why not removed.
|
|
||
| assertEquals(Errors.UNSUPPORTED_VERSION.code, shareFetchResponse.data.errorCode) | ||
| assertEquals(0, shareFetchResponse.data.acquisitionLockTimeoutMs) | ||
|
|
||
| socket.close() |
There was a problem hiding this comment.
So if we avoid this call here and rely on tearDown to close sockets then will it cause any trouble? Then you don't need to have an explicit if (!socket.isClosed) as well?
…owledgeRequestTest
…se the socket for subsequent requests (apache#19640) Currently in the tests in ShareFetchAcknowledgeRequestTest, subsequent share fetch / share acknowledge requests creates a new socket everytime, even when the requests are sent by the same member. In reality, a single share consumer clisnet will reuse the same socket for all the share related requests in its lifetime. This PR changes the behaviour in the tests to align with reality and reuse the same socket for all requests by the same share group member. Reviewers: Apoorv Mittal <apoorvmittal10@gmail.com>
Currently in the tests in ShareFetchAcknowledgeRequestTest, subsequent share fetch / share acknowledge requests creates a new socket everytime, even when the requests are sent by the same member. In reality, a single share consumer clisnet will reuse the same socket for all the share related requests in its lifetime. This PR changes the behaviour in the tests to align with reality and reuse the same socket for all requests by the same share group member.