You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org> on 2025/05/13 07:56:46 UTC
[PR] KAFKA-19231-1: Handle fetch request when share session cache is full [kafka]
chirag-wadhwa5 opened a new pull request, #19701:
URL: https://github.com/apache/kafka/pull/19701
According to the recent changes in KIP-932, when the share session cache is full and a broker receives a Share Fetch request with Initial Share Session Epoch (0), then the error code `SHARE_SESSION_LIMIT_REACHED` is returned after a delay of maxWaitMs. This PR implements this logic. In order to add a delay between subsequent share fetch requests, the timer is delayed operation purgatory is used. A new `IdleShareFetchTimeTask` has been added which takes in a CompletableFuture<Void>. Upon the expiration, this future is completed with null. When the future is completes, a response is sent back to the client with the error code `SHARE_SESSION_LIMIT_REACHED`
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19231-1: Handle fetch request when share session cache is full [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #19701:
URL: https://github.com/apache/kafka/pull/19701#discussion_r2088357326
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
+ val memberId: Uuid = Uuid.ZERO_UUID
+
+ val groupId = "group"
+
+ when(sharePartitionManager.newContext(any(), any(), any(), any(), any(), any()))
+ .thenThrow(Errors.SHARE_SESSION_LIMIT_REACHED.exception)
+
+ when(sharePartitionManager.createIdleShareFetchTimerTask(any()))
+ .thenAnswer(_ => CompletableFuture.runAsync(() => {}))
+
+ val shareFetchRequestData = new ShareFetchRequestData().
+ setGroupId(groupId).
+ setMemberId(memberId.toString).
+ setShareSessionEpoch(0).
+ setTopics(util.List.of(new ShareFetchRequestData.FetchTopic().
+ setTopicId(topicId).
+ setPartitions(util.List.of(
+ new ShareFetchRequestData.FetchPartition()
+ .setPartitionIndex(0)))))
+
+ 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()
+
+ assertEquals(Errors.SHARE_SESSION_LIMIT_REACHED.code, responseData.errorCode)
Review Comment:
Thanks for the review. I have made some edits to the test, which actually makes sure that the response is not returned immediately. Also, the waiting behaviour is tested in SharePartitionManagerTest.testCreateIdleShareFetchTask
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
+ val memberId: Uuid = Uuid.ZERO_UUID
+
+ val groupId = "group"
+
+ when(sharePartitionManager.newContext(any(), any(), any(), any(), any(), any()))
+ .thenThrow(Errors.SHARE_SESSION_LIMIT_REACHED.exception)
+
+ when(sharePartitionManager.createIdleShareFetchTimerTask(any()))
+ .thenAnswer(_ => CompletableFuture.runAsync(() => {}))
Review Comment:
Thanks for the review. I have made edits to the test, removing runAsync altogether
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19231-1: Handle fetch request when share session cache is full [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19701:
URL: https://github.com/apache/kafka/pull/19701#discussion_r2086392228
##########
core/src/main/java/kafka/server/share/IdleShareFetchTimeTask.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server.share;
+
+import org.apache.kafka.server.util.timer.TimerTask;
+
+import java.util.concurrent.CompletableFuture;
+
+public class IdleShareFetchTimeTask extends TimerTask {
Review Comment:
nit: `IdleShareFetchTimerTask` with an extra 'r' please.
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -388,6 +389,21 @@ public CompletableFuture<Map<TopicIdPartition, ShareAcknowledgeResponseData.Part
return mapAcknowledgementFutures(futuresMap, Optional.empty());
}
+ /**
+ * The createShareFetchTask creates a timer task to delay the share fetch request for maxWaitMs duration.
+ * A future is created and returned which will be completed when the timer task is completed.
+ *
+ * @param maxWaitMs The duration after which the timer task will be completed.
+ *
+ * @return A future that will be completed when the timer task is completed.
+ */
+ public CompletableFuture<Void> createIdleShareFetchTask(long maxWaitMs) {
Review Comment:
`createIdleShareFetchTimerTask` perhaps.
##########
core/src/main/java/kafka/server/share/IdleShareFetchTimeTask.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server.share;
+
+import org.apache.kafka.server.util.timer.TimerTask;
+
+import java.util.concurrent.CompletableFuture;
+
+public class IdleShareFetchTimeTask extends TimerTask {
+
+ CompletableFuture<Void> future;
Review Comment:
`private final` would be better.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -448,6 +451,10 @@ class ReplicaManager(val config: KafkaConfig,
delayedShareFetchPurgatory.tryCompleteElseWatch(delayedShareFetch, delayedShareFetchKeys)
}
+ private[server] def addDelayedShareTimerRequest(timerTask: TimerTask): Unit = {
Review Comment:
The naming could be a bit tighter. There are several layers of the similar concepts, but they've all got slightly different names. `addDelayedShareFetchTimerRequest` would be more consistent here I feel.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19231-1: Handle fetch request when share session cache is full [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #19701:
URL: https://github.com/apache/kafka/pull/19701#discussion_r2087080717
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -2803,6 +2804,42 @@ public void testFetchMessagesRotatePartitions() {
validateRotatedListEquals(topicIdPartitions, resultShareFetch.topicIdPartitions(), 1);
}
+ @Test
+ public void testCreateIdleShareFetchTask() throws Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+
+ MockTime mockTime = new MockTime();
Review Comment:
`time` which resets in setup can be used.
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -2803,6 +2804,42 @@ public void testFetchMessagesRotatePartitions() {
validateRotatedListEquals(topicIdPartitions, resultShareFetch.topicIdPartitions(), 1);
}
+ @Test
+ public void testCreateIdleShareFetchTask() throws Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+
+ MockTime mockTime = new MockTime();
+ MockTimer mockTimer = new MockTimer(mockTime);
+ long maxWaitMs = 1000L;
+
+ // Set up the mock to capture and add the timer task
+ Mockito.doAnswer(invocation -> {
+ TimerTask timerTask = invocation.getArgument(0);
+ mockTimer.add(timerTask);
+ return null;
+ }).when(replicaManager).addDelayedShareFetchTimerRequest(Mockito.any(TimerTask.class));
+
+ sharePartitionManager = SharePartitionManagerBuilder.builder()
+ .withReplicaManager(replicaManager)
+ .withTime(mockTime)
+ .withTimer(mockTimer)
+ .build();
+
+ CompletableFuture<Void> future = sharePartitionManager.createIdleShareFetchTimerTask(maxWaitMs);
+
+ // Future should not be completed immediately
+ assertFalse(future.isDone());
+
+ mockTimer.advanceClock(maxWaitMs / 2);
+ assertFalse(future.isDone());
+
+ mockTimer.advanceClock(maxWaitMs);
+
Review Comment:
nit: remove line break
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
+ val memberId: Uuid = Uuid.ZERO_UUID
Review Comment:
Can be inline, not required here.
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
+ val memberId: Uuid = Uuid.ZERO_UUID
+
+ val groupId = "group"
+
+ when(sharePartitionManager.newContext(any(), any(), any(), any(), any(), any()))
+ .thenThrow(Errors.SHARE_SESSION_LIMIT_REACHED.exception)
+
+ when(sharePartitionManager.createIdleShareFetchTimerTask(any()))
+ .thenAnswer(_ => CompletableFuture.runAsync(() => {}))
+
+ val shareFetchRequestData = new ShareFetchRequestData().
+ setGroupId(groupId).
Review Comment:
```suggestion
setGroupId("group").
```
##########
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/ShareConsumerTest.java:
##########
@@ -2199,6 +2199,41 @@ public void testShareGroupMaxSizeConfigExceeded() throws Exception {
shareConsumer4.close();
}
+ @ClusterTest(
+ brokers = 1,
+ serverProperties = {
+ @ClusterConfigProperty(key = "group.share.max.size", value = "1"), // Setting max group size to 1
+ @ClusterConfigProperty(key = "group.share.max.share.sessions", value = "1") // Setting max share sessions value to 1
+ }
+ )
+ public void testShareGroupShareSessionCacheIsFull() {
+ alterShareAutoOffsetReset("group1", "earliest");
+ try (Producer<byte[], byte[]> producer = createProducer();
+ ShareConsumer<byte[], byte[]> shareConsumer1 = createShareConsumer("group1");
+ ShareConsumer<byte[], byte[]> shareConsumer2 = createShareConsumer("group2")) {
+
+ ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(tp.topic(), tp.partition(), null, "key".getBytes(), "value".getBytes());
+ producer.send(record);
+ producer.flush();
+ shareConsumer1.subscribe(Set.of(tp.topic()));
+ shareConsumer2.subscribe(Set.of(tp.topic()));
+
+ ConsumerRecords<byte[], byte[]> records = waitedPoll(shareConsumer1, 2500L, 1);
+ assertEquals(1, records.count());
+
+ producer.send(record);
+ producer.flush();
+
+ // The second share consumer should not throw any exception, but should not receive any records as well.
+ records = shareConsumer2.poll(Duration.ofMillis(1000));
Review Comment:
So the client received `session error` but didn't provide records to application, correct? Does there appear log line in client which specifies the session error?
##########
core/src/main/java/kafka/server/share/IdleShareFetchTimerTask.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server.share;
Review Comment:
Also can be an inner class in SharePartitionManager.
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
Review Comment:
```suggestion
addTopicToMetadataCache("foo", 1, topicId = topicId)
```
##########
core/src/main/java/kafka/server/share/IdleShareFetchTimerTask.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server.share;
+
+import org.apache.kafka.server.util.timer.TimerTask;
+
+import java.util.concurrent.CompletableFuture;
+
+public class IdleShareFetchTimerTask extends TimerTask {
Review Comment:
Need class java docs.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -448,6 +451,10 @@ class ReplicaManager(val config: KafkaConfig,
delayedShareFetchPurgatory.tryCompleteElseWatch(delayedShareFetch, delayedShareFetchKeys)
}
+ private[server] def addDelayedShareFetchTimerRequest(timerTask: TimerTask): Unit = {
Review Comment:
Need method comments.
##########
core/src/test/java/kafka/server/share/SharePartitionManagerTest.java:
##########
@@ -2803,6 +2804,42 @@ public void testFetchMessagesRotatePartitions() {
validateRotatedListEquals(topicIdPartitions, resultShareFetch.topicIdPartitions(), 1);
}
+ @Test
+ public void testCreateIdleShareFetchTask() throws Exception {
+ ReplicaManager replicaManager = mock(ReplicaManager.class);
+
+ MockTime mockTime = new MockTime();
+ MockTimer mockTimer = new MockTimer(mockTime);
+ long maxWaitMs = 1000L;
+
+ // Set up the mock to capture and add the timer task
+ Mockito.doAnswer(invocation -> {
+ TimerTask timerTask = invocation.getArgument(0);
+ mockTimer.add(timerTask);
+ return null;
+ }).when(replicaManager).addDelayedShareFetchTimerRequest(Mockito.any(TimerTask.class));
+
+ sharePartitionManager = SharePartitionManagerBuilder.builder()
+ .withReplicaManager(replicaManager)
+ .withTime(mockTime)
+ .withTimer(mockTimer)
+ .build();
+
+ CompletableFuture<Void> future = sharePartitionManager.createIdleShareFetchTimerTask(maxWaitMs);
+
Review Comment:
nit: remove line break.
##########
core/src/main/java/kafka/server/share/IdleShareFetchTimerTask.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server.share;
Review Comment:
Should not be in `core` but in `server` module.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -291,6 +292,8 @@ class ReplicaManager(val config: KafkaConfig,
private val metricsGroup = new KafkaMetricsGroup(this.getClass)
private val addPartitionsToTxnConfig = new AddPartitionsToTxnConfig(config)
Review Comment:
nit: remove line break.
##########
core/src/main/java/kafka/server/share/SharePartitionManager.java:
##########
@@ -388,6 +389,21 @@ public CompletableFuture<Map<TopicIdPartition, ShareAcknowledgeResponseData.Part
return mapAcknowledgementFutures(futuresMap, Optional.empty());
}
+ /**
+ * The createShareFetchTask creates a timer task to delay the share fetch request for maxWaitMs duration.
Review Comment:
```suggestion
* The method creates a timer task to delay the share fetch request for maxWaitMs duration.
```
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
+ val memberId: Uuid = Uuid.ZERO_UUID
+
+ val groupId = "group"
+
+ when(sharePartitionManager.newContext(any(), any(), any(), any(), any(), any()))
+ .thenThrow(Errors.SHARE_SESSION_LIMIT_REACHED.exception)
+
+ when(sharePartitionManager.createIdleShareFetchTimerTask(any()))
+ .thenAnswer(_ => CompletableFuture.runAsync(() => {}))
+
+ val shareFetchRequestData = new ShareFetchRequestData().
+ setGroupId(groupId).
+ setMemberId(memberId.toString).
Review Comment:
```suggestion
setMemberId(Uuid.randomUuid.toString).
```
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
+ val memberId: Uuid = Uuid.ZERO_UUID
+
+ val groupId = "group"
+
+ when(sharePartitionManager.newContext(any(), any(), any(), any(), any(), any()))
+ .thenThrow(Errors.SHARE_SESSION_LIMIT_REACHED.exception)
+
+ when(sharePartitionManager.createIdleShareFetchTimerTask(any()))
+ .thenAnswer(_ => CompletableFuture.runAsync(() => {}))
Review Comment:
Query: why do we need runAsync?
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
+ val memberId: Uuid = Uuid.ZERO_UUID
+
+ val groupId = "group"
+
+ when(sharePartitionManager.newContext(any(), any(), any(), any(), any(), any()))
+ .thenThrow(Errors.SHARE_SESSION_LIMIT_REACHED.exception)
+
+ when(sharePartitionManager.createIdleShareFetchTimerTask(any()))
+ .thenAnswer(_ => CompletableFuture.runAsync(() => {}))
+
+ val shareFetchRequestData = new ShareFetchRequestData().
+ setGroupId(groupId).
+ setMemberId(memberId.toString).
+ setShareSessionEpoch(0).
+ setTopics(util.List.of(new ShareFetchRequestData.FetchTopic().
+ setTopicId(topicId).
+ setPartitions(util.List.of(
+ new ShareFetchRequestData.FetchPartition()
+ .setPartitionIndex(0)))))
+
+ 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()
+
+ assertEquals(Errors.SHARE_SESSION_LIMIT_REACHED.code, responseData.errorCode)
Review Comment:
How do we verify the if request waited for maxWaitTime?
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
+ val topicId = Uuid.randomUuid()
+ metadataCache = initializeMetadataCacheWithShareGroupsEnabled()
+ addTopicToMetadataCache(topicName, 1, topicId = topicId)
+ val memberId: Uuid = Uuid.ZERO_UUID
+
+ val groupId = "group"
Review Comment:
Can be inline, not required here.
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -4824,6 +4824,42 @@ class KafkaApisTest extends Logging {
assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, responseData.errorCode)
}
+ @Test
+ def testHandleShareFetchRequestWhenShareSessionCacheIsFull(): Unit = {
+ val topicName = "foo"
Review Comment:
Can be inline, not required here.
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -448,6 +451,10 @@ class ReplicaManager(val config: KafkaConfig,
delayedShareFetchPurgatory.tryCompleteElseWatch(delayedShareFetch, delayedShareFetchKeys)
}
+ private[server] def addDelayedShareFetchTimerRequest(timerTask: TimerTask): Unit = {
Review Comment:
addDelayedShareFetchTimerRequest => addShareFetchTimerRequest
##########
core/src/main/java/kafka/server/share/IdleShareFetchTimerTask.java:
##########
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package kafka.server.share;
+
+import org.apache.kafka.server.util.timer.TimerTask;
+
+import java.util.concurrent.CompletableFuture;
+
+public class IdleShareFetchTimerTask extends TimerTask {
+
+ private final CompletableFuture<Void> future;
Review Comment:
Need comments.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org