chirag-wadhwa5 commented on code in PR #19637:
URL: https://github.com/apache/kafka/pull/19637#discussion_r2077345638
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -6380,6 +6380,42 @@ class KafkaApisTest extends Logging {
assertEquals(0,
topicResponses.get(0).partitions.get(0).acquiredRecords.toArray().length)
}
+ @Test
+ def handleShareFetchRequestThrottlingWhenSessionCacheIsFull(): Unit = {
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+
+ val groupId = "group"
+ val memberId: Uuid = Uuid.ZERO_UUID
+ val maxWaitMs: Int = 2000
+
+ doThrow(new ShareSessionLimitReachedException("Share session exceeded"))
+ .when(sharePartitionManager).newContext(any(), any(), any(), any(),
any(), any());
+
+ val shareFetchRequestData = new ShareFetchRequestData().
+ setGroupId(groupId).
+ setMemberId(memberId.toString).
+ setShareSessionEpoch(ShareRequestMetadata.INITIAL_EPOCH).
+ setTopics(List().asJava).
+ setMaxWaitMs(maxWaitMs)
+
+ val shareFetchRequest = new
ShareFetchRequest.Builder(shareFetchRequestData).build(ApiKeys.SHARE_FETCH.latestVersion)
+ val request = buildRequest(shareFetchRequest)
+ kafkaApis = createKafkaApis()
+ kafkaApis.handleShareFetchRequest(request)
+ val response = verifyNoThrottling[ShareFetchResponse](request)
+ val responseData = response.data()
+
+ val expectedThrottleTimeMs = maxWaitMs
+
+ verify(clientRequestQuotaManager).throttle(
+ ArgumentMatchers.eq(request),
+ any[ThrottleCallback](),
+ ArgumentMatchers.eq(expectedThrottleTimeMs)
+ )
+
+ assertEquals(expectedThrottleTimeMs, responseData.throttleTimeMs)
+ }
Review Comment:
Thanks for the review.
I don't think I can extend the test in handleShareFetchRequest because in
that file, we don't make use of any channels or sockets. We simply call the
handleShareFetchRequest method. But the throttling logic is implemented on the
channel / socket level.
I do have a test prepared for ShareFetchAckowledgeRequestTest file, but that
would require successive requests to use the same sockets so that they can be
identifies as sent from the same share consumer client. That would be dependent
on the PR https://github.com/apache/kafka/pull/19640. As soon as the PR gets
merged, I will update this PR with the test.
--
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]
