TaiJuWu commented on PR #19105: URL: https://github.com/apache/kafka/pull/19105#issuecomment-2708878915
> > If the request number exceeds maxInFlightRequestsPerConnection, the LeaveGroup request would be not sent in time when closing. > > why does it exceed the `maxInFlightRequestsPerConnection`? Does it happen in the test only? Base on log, follow requests are sent: ``` [2025-01-14 11:01:57,082] DEBUG [Consumer clientId=ConsumerTestConsumer, groupId=my-test] kkk ClientRequest(expectResponse=true, callback=org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler@1af677f8, destination=-1, correlationId=0, clientId=ConsumerTestConsumer, createdTimeMs=1736852517076, requestBuilder=FindCoordinatorRequestData(key='my-test', keyType=0, coordinatorKeys=[])) (org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient:516) [2025-01-14 11:01:57,100] DEBUG [Consumer clientId=ConsumerTestConsumer, groupId=my-test] kkk ClientRequest(expectResponse=true, callback=org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler@28cb86b2, destination=2147483646, correlationId=5, clientId=ConsumerTestConsumer, createdTimeMs=1736852517096, requestBuilder=JoinGroupRequestData(groupId='my-test', sessionTimeoutMs=45000, rebalanceTimeoutMs=6000, memberId='', groupInstanceId=null, protocolType='consumer', protocols=[JoinGroupRequestProtocol(name='range', metadata=[0, 3, 0, 0, 0, 1, 0, 5, 116, 111, 112, 105, 99, -1, -1, -1, -1, 0, 0, 0, 0, -1, -1, -1, -1, -1, -1]), JoinGroupRequestProtocol(name='cooperative-sticky', metadata=[0, 3, 0, 0, 0, 1, 0, 5, 116, 111, 112, 105, 99, 0, 0, 0, 4, -1, -1, -1, -1, 0, 0, 0, 0, -1, -1, -1, -1, -1, -1])], reason='')) (org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient:516) [2025-01-14 11:01:57,109] DEBUG [Consumer clientId=ConsumerTestConsumer, groupId=my-test] kkk ClientRequest(expectResponse=true, callback=org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler@19705650, destination=2147483646, correlationId=7, clientId=ConsumerTestConsumer, createdTimeMs=1736852517109, requestBuilder=JoinGroupRequestData(groupId='my-test', sessionTimeoutMs=45000, rebalanceTimeoutMs=6000, memberId='ConsumerTestConsumer-2034ef08-671b-4849-a8f4-fffa51ab3d28', groupInstanceId=null, protocolType='consumer', protocols=[JoinGroupRequestProtocol(name='range', metadata=[0, 3, 0, 0, 0, 1, 0, 5, 116, 111, 112, 105, 99, -1, -1, -1, -1, 0, 0, 0, 0, -1, -1, -1, -1, -1, -1]), JoinGroupRequestProtocol(name='cooperative-sticky', metadata=[0, 3, 0, 0, 0, 1, 0, 5, 116, 111, 112, 105, 99, 0, 0, 0, 4, -1, -1, -1, -1, 0, 0, 0, 0, -1, -1, -1, -1, -1, -1])], reason='need to re-join with the given member-id: ConsumerTestConsumer-2034ef08-671b-4849 -a8f4-fffa51ab3d28')) (org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient:516) [2025-01-14 11:01:57,145] DEBUG [Consumer clientId=ConsumerTestConsumer, groupId=my-test] kkk ClientRequest(expectResponse=true, callback=org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler@1a76202b, destination=2147483646, correlationId=8, clientId=ConsumerTestConsumer, createdTimeMs=1736852517144, requestBuilder=SyncGroupRequestData(groupId='my-test', generationId=1, memberId='ConsumerTestConsumer-2034ef08-671b-4849-a8f4-fffa51ab3d28', groupInstanceId=null, protocolType='consumer', protocolName='range', assignments=[SyncGroupRequestAssignment(memberId='ConsumerTestConsumer-2034ef08-671b-4849-a8f4-fffa51ab3d28', assignment=[0, 3, 0, 0, 0, 1, 0, 5, 116, 111, 112, 105, 99, 0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 1, -1, -1, -1, -1])])) (org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient:516) [2025-01-14 11:01:57,168] DEBUG [Consumer clientId=ConsumerTestConsumer, groupId=my-test] kkk ClientRequest(expectResponse=true, callback=org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler@74b1838, destination=2147483646, correlationId=9, clientId=ConsumerTestConsumer, createdTimeMs=1736852517166, requestBuilder=OffsetFetchRequestData(groupId='', topics=[], groups=[OffsetFetchRequestGroup(groupId='my-test', memberId=null, memberEpoch=-1, topics=[OffsetFetchRequestTopics(name='topic', partitionIndexes=[0, 1])])], requireStable=true)) (org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient:516) ``` > Does it happen in the test only? If I didn't miss anything, I did not see other integration test related `consumer#close`. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
