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 2024/08/19 10:27:29 UTC
[PR] KAFKA-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
chirag-wadhwa5 opened a new pull request, #16916:
URL: https://github.com/apache/kafka/pull/16916
This PR contains Integration Tests for Share Fetch and Share Acknowledge APIs
Reference: [KAFKA-17290](https://issues.apache.org/jira/projects/KAFKA/issues/KAFKA-17290?filter=allopenissues)
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1724670046
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -25,48 +25,14 @@ import org.apache.kafka.common.message.LeaveGroupResponseData.MemberResponse
import org.apache.kafka.common.message.SyncGroupRequestData.SyncGroupRequestAssignment
import org.apache.kafka.common.message.{ConsumerGroupDescribeRequestData, ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsRequestData, DeleteGroupsResponseData, DescribeGroupsRequestData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, OffsetFetchResponseData, ShareGroupDescribeRequestData, ShareGroupDescribeResponseData, ShareGroupHeartbeatRequestData, ShareGroupHeartbeatResponseData, SyncGroupRequestData, SyncGroupResponseData}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse, DescribeGroupsRequest, DescribeGroupsResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsRequest, ListGroupsResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetDeleteRequest, OffsetDeleteResponse, OffsetFetchRequest, OffsetFetchResponse, ShareGroupDescribeRequest, ShareGroupDescribeResponse, ShareGroupHeartbeatRequest, ShareGroupHeartbeatResponse, SyncGroupRequest, SyncGroupResponse}
+import org.apache.kafka.common.requests.{ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse, DescribeGroupsRequest, DescribeGroupsResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsRequest, ListGroupsResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetDeleteRequest, OffsetDeleteResponse, OffsetFetchRequest, OffsetFetchResponse, ShareGroupDescribeRequest, ShareGroupDescribeResponse, ShareGroupHeartbeatRequest, ShareGroupHeartbeatResponse, SyncGroupRequest, SyncGroupResponse}
import org.junit.jupiter.api.Assertions.{assertEquals, fail}
+import unit.kafka.server.RequestUtilitiesTest
import java.util.Comparator
-import java.util.stream.Collectors
import scala.jdk.CollectionConverters._
-import scala.reflect.ClassTag
-class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
- private def brokers(): Seq[KafkaBroker] = cluster.brokers.values().stream().collect(Collectors.toList[KafkaBroker]).asScala.toSeq
-
- private def controllerServers(): Seq[ControllerServer] = cluster.controllers().values().asScala.toSeq
-
- protected def createOffsetsTopic(): Unit = {
- TestUtils.createOffsetsTopicWithAdmin(
- admin = cluster.createAdminClient(),
- brokers = brokers(),
- controllers = controllerServers()
- )
- }
-
- protected def createTopic(
- topic: String,
- numPartitions: Int
- ): Unit = {
- TestUtils.createTopicWithAdmin(
- admin = cluster.createAdminClient(),
- brokers = brokers(),
- controllers = controllerServers(),
- topic = topic,
- numPartitions = numPartitions
- )
- }
-
- protected def isUnstableApiEnabled: Boolean = {
- cluster.config.serverProperties.get("unstable.api.versions.enable") == "true"
- }
-
- protected def isNewGroupCoordinatorEnabled: Boolean = {
- cluster.config.serverProperties.get("group.coordinator.new.enable") == "true" ||
- cluster.config.serverProperties.get("group.coordinator.rebalance.protocols").contains("consumer")
- }
+class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) extends RequestUtilitiesTest(cluster) {
Review Comment:
Why can't it be a composition and RequestUtilities can be instatiated with cluster instance in test class itself?
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1749568819
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -38,6 +46,12 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
private def controllerServers(): Seq[ControllerServer] = cluster.controllers().values().asScala.toSeq
+ protected def securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT
Review Comment:
we are always passing securityProtocol, listenerName. looks these are not necessary here
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1724505627
##########
core/src/test/scala/unit/kafka/server/RequestUtilitiesTest.scala:
##########
@@ -0,0 +1,152 @@
+/**
+ * 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 unit.kafka.server
+
+import kafka.network.SocketServer
+import kafka.server.{ControllerServer, IntegrationTestUtils, KafkaBroker}
+import kafka.test.ClusterInstance
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse}
+import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.serialization.StringSerializer
+import org.apache.kafka.controller.ControllerRequestContextUtil.ANONYMOUS_CONTEXT
+import org.junit.jupiter.api.AfterEach
+
+import java.util.Properties
+import java.util.stream.Collectors
+import scala.collection.Seq
+import scala.collection.convert.ImplicitConversions.{`collection AsScalaIterable`, `map AsScala`}
+import scala.reflect.ClassTag
+
+class RequestUtilitiesTest(cluster: ClusterInstance) {
Review Comment:
Thanks for the review, ```RequestUtilities``` seems to be a better name. I will make the change in the next commit
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy merged PR #16916:
URL: https://github.com/apache/kafka/pull/16916
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1747537854
##########
core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala:
##########
@@ -0,0 +1,1863 @@
+/**
+ * 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 unit.kafka.server
Review Comment:
package unit.kafka.server => package kafka.server
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1742713694
##########
core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala:
##########
@@ -0,0 +1,1861 @@
+/**
+ * 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 unit.kafka.server
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords
+import org.apache.kafka.common.message.{ShareAcknowledgeRequestData, ShareAcknowledgeResponseData, ShareFetchRequestData, ShareFetchResponseData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.common.requests.{ShareAcknowledgeRequest, ShareAcknowledgeResponse, ShareFetchRequest, ShareFetchResponse, ShareRequestMetadata}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.{AfterEach, Tag, Timeout}
+import org.junit.jupiter.api.extension.ExtendWith
+
+import java.util
+import java.util.Collections
+import scala.collection.convert.ImplicitConversions.`list asScalaBuffer`
+import scala.jdk.CollectionConverters._
+
+@Timeout(120)
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(types = Array(Type.KRAFT), brokers = 1)
+@Tag("integration")
+class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance){
+
+ @AfterEach
+ def tearDown(): Unit = {
+ requestUtilities.closeProducer()
+ }
+
+ private val requestUtilities: RequestUtilities = new RequestUtilities(cluster)
+ private final val MAX_PARTITION_BYTES = 10000
Review Comment:
Should we have these variables/constants before any method?
##########
core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala:
##########
@@ -0,0 +1,1861 @@
+/**
+ * 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 unit.kafka.server
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords
Review Comment:
nit: a line break
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -76,7 +45,7 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
offset: Long,
expectedError: Errors,
version: Short
- ): Unit = {
+ ): Unit = {
Review Comment:
Is this change needed?
##########
core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala:
##########
@@ -0,0 +1,1861 @@
+/**
+ * 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 unit.kafka.server
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords
+import org.apache.kafka.common.message.{ShareAcknowledgeRequestData, ShareAcknowledgeResponseData, ShareFetchRequestData, ShareFetchResponseData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.common.requests.{ShareAcknowledgeRequest, ShareAcknowledgeResponse, ShareFetchRequest, ShareFetchResponse, ShareRequestMetadata}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.{AfterEach, Tag, Timeout}
+import org.junit.jupiter.api.extension.ExtendWith
+
+import java.util
+import java.util.Collections
+import scala.collection.convert.ImplicitConversions.`list asScalaBuffer`
Review Comment:
nit: a line break
##########
core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala:
##########
@@ -0,0 +1,1861 @@
+/**
+ * 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 unit.kafka.server
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords
+import org.apache.kafka.common.message.{ShareAcknowledgeRequestData, ShareAcknowledgeResponseData, ShareFetchRequestData, ShareFetchResponseData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.common.requests.{ShareAcknowledgeRequest, ShareAcknowledgeResponse, ShareFetchRequest, ShareFetchResponse, ShareRequestMetadata}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.{AfterEach, Tag, Timeout}
+import org.junit.jupiter.api.extension.ExtendWith
+
+import java.util
+import java.util.Collections
+import scala.collection.convert.ImplicitConversions.`list asScalaBuffer`
+import scala.jdk.CollectionConverters._
+
+@Timeout(120)
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(types = Array(Type.KRAFT), brokers = 1)
+@Tag("integration")
+class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance){
+
+ @AfterEach
+ def tearDown(): Unit = {
+ requestUtilities.closeProducer()
Review Comment:
Should it be just `requestUtilities.closeProducer`?
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "omkreddy (via GitHub)" <gi...@apache.org>.
omkreddy commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1749568819
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -38,6 +46,12 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
private def controllerServers(): Seq[ControllerServer] = cluster.controllers().values().asScala.toSeq
+ protected def securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT
Review Comment:
we are always passing securityProtocol, listenerName. looks the are not necessary here
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -67,6 +99,56 @@ class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
cluster.brokers.values.stream.allMatch(b => b.config.isNewGroupCoordinatorEnabled)
}
+ protected def getTopicIds: Map[String, Uuid] = {
+ cluster.controllers().get(cluster.controllerIds().iterator().next()).controller.findAllTopicIds(ANONYMOUS_CONTEXT).get().toMap
+ }
+
+ protected def getBrokers: Seq[KafkaBroker] = {
+ cluster.brokers.values().stream().collect(Collectors.toList[KafkaBroker]).toSeq
+ }
+
+ protected def brokerSocketServer(brokerId: Int): SocketServer = {
Review Comment:
This can be private
##########
core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala:
##########
@@ -0,0 +1,1861 @@
+/**
+ * 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
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+
+import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords
+import org.apache.kafka.common.message.{ShareAcknowledgeRequestData, ShareAcknowledgeResponseData, ShareFetchRequestData, ShareFetchResponseData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.common.requests.{ShareAcknowledgeRequest, ShareAcknowledgeResponse, ShareFetchRequest, ShareFetchResponse, ShareRequestMetadata}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.{AfterEach, Tag, Timeout}
+import org.junit.jupiter.api.extension.ExtendWith
+
+import java.util
+import java.util.Collections
+import scala.collection.convert.ImplicitConversions.`list asScalaBuffer`
+import scala.jdk.CollectionConverters._
+
+@Timeout(120)
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(types = Array(Type.KRAFT), brokers = 1)
+@Tag("integration")
+class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster){
+
+ private final val MAX_PARTITION_BYTES = 10000
+
+ @AfterEach
+ def tearDown(): Unit = {
+ closeProducer
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true")
+ )
+ )
+ def testShareFetchRequestIsInAccessibleWhenConfigsDisabled(): Unit = {
+ val groupId: String = "group"
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(Uuid.randomUuid(), ShareRequestMetadata.INITIAL_EPOCH)
+ val send: Seq[TopicIdPartition] = Seq(
+ new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic1", 0)),
+ new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic1", 1))
+ )
+
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ assertEquals(Errors.UNSUPPORTED_VERSION.code(), shareFetchResponse.data().errorCode())
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true")
+ )
+ )
+ def testShareAcknowledgeRequestIsInAccessibleWhenConfigsDisabled(): Unit = {
+ val groupId: String = "group"
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(Uuid.randomUuid(), ShareRequestMetadata.INITIAL_EPOCH)
+
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, Map.empty)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ assertEquals(Errors.UNSUPPORTED_VERSION.code(), shareAcknowledgeResponse.data().errorCode())
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ ),
+ brokers = 2
+ )
+ def testShareFetchRequestToNonReplica(): Unit = {
Review Comment:
NotLeader => NonLeaderReplica
##########
core/src/test/scala/unit/kafka/server/ShareFetchAcknowledgeRequestTest.scala:
##########
@@ -0,0 +1,1861 @@
+/**
+ * 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
+
+import kafka.test.ClusterInstance
+import kafka.test.annotation.{ClusterConfigProperty, ClusterTest, ClusterTestDefaults, Type}
+import kafka.test.junit.ClusterTestExtensions
+
+import org.apache.kafka.common.message.ShareFetchResponseData.AcquiredRecords
+import org.apache.kafka.common.message.{ShareAcknowledgeRequestData, ShareAcknowledgeResponseData, ShareFetchRequestData, ShareFetchResponseData}
+import org.apache.kafka.common.protocol.Errors
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.common.requests.{ShareAcknowledgeRequest, ShareAcknowledgeResponse, ShareFetchRequest, ShareFetchResponse, ShareRequestMetadata}
+import org.junit.jupiter.api.Assertions.{assertEquals, assertTrue}
+import org.junit.jupiter.api.{AfterEach, Tag, Timeout}
+import org.junit.jupiter.api.extension.ExtendWith
+
+import java.util
+import java.util.Collections
+import scala.collection.convert.ImplicitConversions.`list asScalaBuffer`
+import scala.jdk.CollectionConverters._
+
+@Timeout(120)
+@ExtendWith(value = Array(classOf[ClusterTestExtensions]))
+@ClusterTestDefaults(types = Array(Type.KRAFT), brokers = 1)
+@Tag("integration")
+class ShareFetchAcknowledgeRequestTest(cluster: ClusterInstance) extends GroupCoordinatorBaseRequestTest(cluster){
+
+ private final val MAX_PARTITION_BYTES = 10000
+
+ @AfterEach
+ def tearDown(): Unit = {
+ closeProducer
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true")
+ )
+ )
+ def testShareFetchRequestIsInAccessibleWhenConfigsDisabled(): Unit = {
+ val groupId: String = "group"
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(Uuid.randomUuid(), ShareRequestMetadata.INITIAL_EPOCH)
+ val send: Seq[TopicIdPartition] = Seq(
+ new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic1", 0)),
+ new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("topic1", 1))
+ )
+
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ assertEquals(Errors.UNSUPPORTED_VERSION.code(), shareFetchResponse.data().errorCode())
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "unstable.api.versions.enable", value = "true")
+ )
+ )
+ def testShareAcknowledgeRequestIsInAccessibleWhenConfigsDisabled(): Unit = {
+ val groupId: String = "group"
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(Uuid.randomUuid(), ShareRequestMetadata.INITIAL_EPOCH)
+
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, Map.empty)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ assertEquals(Errors.UNSUPPORTED_VERSION.code(), shareAcknowledgeResponse.data().errorCode())
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ ),
+ brokers = 2
+ )
+ def testShareFetchRequestToNonReplica(): Unit = {
+ val groupId: String = "group"
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(Uuid.randomUuid(), ShareRequestMetadata.INITIAL_EPOCH)
+
+ val topic = "topic"
+ val partition = 0
+
+ // Create a single-partition topic and find a broker which is not the leader
+ val partitionToLeader = createTopicAndReturnLeaders(topic)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+ val topicNames = topicIds.asScala.map(_.swap).asJava
+ val leader = partitionToLeader(topicIdPartition)
+ val nonReplicaOpt = getBrokers.find(_.config.brokerId != leader)
+ assertTrue(nonReplicaOpt.isDefined)
+ val nonReplicaId = nonReplicaOpt.get.config.brokerId
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to the non-replica and verify the error code
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest, nonReplicaId)
+ val partitionData = shareFetchResponse.responseData(topicNames).get(topicIdPartition)
+ assertEquals(Errors.NOT_LEADER_OR_FOLLOWER.code, partitionData.errorCode)
+ assertEquals(leader, partitionData.currentLeader().leaderId())
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestSuccess(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ val shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ val expectedPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val partitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(partitionData, expectedPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestSuccessMultiplePartitions(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition1 = new TopicIdPartition(topicId, new TopicPartition(topic, 0))
+ val topicIdPartition2 = new TopicIdPartition(topicId, new TopicPartition(topic, 1))
+ val topicIdPartition3 = new TopicIdPartition(topicId, new TopicPartition(topic, 2))
+
+ initProducer()
+ // Producing 10 records to the topic partitions created above
+ produceData(topicIdPartition1, 10)
+ produceData(topicIdPartition2, 10)
+ produceData(topicIdPartition3, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition1, topicIdPartition2, topicIdPartition3)
+
+ // Send the share fetch request to fetch the records produced above
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ val shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(3, shareFetchResponseData.responses().get(0).partitions().size())
+
+ val expectedPartitionData1 = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(0)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val expectedPartitionData2 = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(1)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val expectedPartitionData3 = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(2)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ shareFetchResponseData.responses().get(0).partitions().foreach(partitionData => {
+ partitionData.partitionIndex() match {
+ case 0 => compareFetchResponsePartitions(partitionData, expectedPartitionData1)
+ case 1 => compareFetchResponsePartitions(partitionData, expectedPartitionData2)
+ case 2 => compareFetchResponsePartitions(partitionData, expectedPartitionData3)
+ }
+ })
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ ),
+ brokers = 3
+ )
+ def testShareFetchRequestSuccessMultiplePartitionsMultipleBrokers(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+
+ val partitionToLeaders = createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition1 = new TopicIdPartition(topicId, new TopicPartition(topic, 0))
+ val topicIdPartition2 = new TopicIdPartition(topicId, new TopicPartition(topic, 1))
+ val topicIdPartition3 = new TopicIdPartition(topicId, new TopicPartition(topic, 2))
+
+ val leader1 = partitionToLeaders(topicIdPartition1)
+ val leader2 = partitionToLeaders(topicIdPartition2)
+ val leader3 = partitionToLeaders(topicIdPartition3)
+
+ initProducer()
+ // Producing 10 records to the topic partitions created above
+ produceData(topicIdPartition1, 10)
+ produceData(topicIdPartition2, 10)
+ produceData(topicIdPartition3, 10)
+
+ val send1: Seq[TopicIdPartition] = Seq(topicIdPartition1)
+ val send2: Seq[TopicIdPartition] = Seq(topicIdPartition2)
+ val send3: Seq[TopicIdPartition] = Seq(topicIdPartition3)
+
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+
+ // Crete different share fetch requests for different partitions as they may have leaders on separate brokers
+ val shareFetchRequest1 = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send1, Seq.empty, acknowledgementsMap)
+ val shareFetchRequest2 = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send2, Seq.empty, acknowledgementsMap)
+ val shareFetchRequest3 = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send3, Seq.empty, acknowledgementsMap)
+
+ val shareFetchResponse1 = connectAndReceive[ShareFetchResponse](shareFetchRequest1, destination = leader1)
+ val shareFetchResponse2 = connectAndReceive[ShareFetchResponse](shareFetchRequest2, destination = leader2)
+ val shareFetchResponse3 = connectAndReceive[ShareFetchResponse](shareFetchRequest3, destination = leader3)
+
+ val shareFetchResponseData1 = shareFetchResponse1.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData1.errorCode)
+ assertEquals(1, shareFetchResponseData1.responses().size())
+ assertEquals(topicId, shareFetchResponseData1.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData1.responses().get(0).partitions().size())
+ val partitionData1 = shareFetchResponseData1.responses().get(0).partitions().get(0)
+
+ val shareFetchResponseData2 = shareFetchResponse2.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData2.errorCode)
+ assertEquals(1, shareFetchResponseData2.responses().size())
+ assertEquals(topicId, shareFetchResponseData2.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData2.responses().get(0).partitions().size())
+ val partitionData2 = shareFetchResponseData2.responses().get(0).partitions().get(0)
+
+ val shareFetchResponseData3 = shareFetchResponse3.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData3.errorCode)
+ assertEquals(1, shareFetchResponseData3.responses().size())
+ assertEquals(topicId, shareFetchResponseData3.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData3.responses().get(0).partitions().size())
+ val partitionData3 = shareFetchResponseData3.responses().get(0).partitions().get(0)
+
+ val expectedPartitionData1 = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(0)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val expectedPartitionData2 = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(1)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val expectedPartitionData3 = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(2)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ compareFetchResponsePartitions(partitionData1, expectedPartitionData1)
+ compareFetchResponsePartitions(partitionData2, expectedPartitionData2)
+ compareFetchResponsePartitions(partitionData3, expectedPartitionData3)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareAcknowledgeRequestSuccessAccept(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ val acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Send a Share Acknowledge request to acknowledge the fetched records
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ val acknowledgementsMapForAcknowledge: Map[TopicIdPartition, util.List[ShareAcknowledgeRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition -> List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava) // Accept the records
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMapForAcknowledge)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.NONE.code, shareAcknowledgeResponseData.errorCode)
+ assertEquals(1, shareAcknowledgeResponseData.responses().size())
+ assertEquals(topicId, shareAcknowledgeResponseData.responses().get(0).topicId())
+ assertEquals(1, shareAcknowledgeResponseData.responses().get(0).partitions().size())
+
+ val expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+
+ val acknowledgePartitionData = shareAcknowledgeResponseData.responses().get(0).partitions().get(0)
+ compareAcknowledgeResponsePartitions(acknowledgePartitionData, expectedAcknowledgePartitionData)
+
+ // Producing 10 more records to the topic
+ produceData(topicIdPartition, 10)
+
+ // Sending a second share fetch request to check if acknowledgements were done successfully
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(10), Collections.singletonList(19), Collections.singletonList(1))) // Only the records from offset 10 onwards should be fetched because records at offsets 0-9 have been acknowledged
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestPiggybackedAccept(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ var acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Producing 10 more records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ // Send a Share Fetch request with piggybacked acknowledgements
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ acknowledgementsMapForFetch = Map(topicIdPartition -> List(new ShareFetchRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava) // Accept the records
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(10), Collections.singletonList(19), Collections.singletonList(1))) // The records at offsets 0 to 9 will not be re fetched because they have been acknowledged
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Producing 10 more records to the topic
+ produceData(topicIdPartition, 10)
+
+ // Sending a third share fetch request to confirm if acknowledgements were done successfully
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(20), Collections.singletonList(29), Collections.singletonList(1))) // Only the records from offset 20 onwards should be fetched because records at offsets 0-9 have been acknowledged before and 10 to 19 are currently acquired
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareAcknowledgeRequestSuccessRelease(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ val acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Send a Share Acknowledge request to acknowledge the fetched records
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ val acknowledgementsMapForAcknowledge: Map[TopicIdPartition, util.List[ShareAcknowledgeRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition -> List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(2.toByte))).asJava) // Release the records
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMapForAcknowledge)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.NONE.code, shareAcknowledgeResponseData.errorCode)
+ assertEquals(1, shareAcknowledgeResponseData.responses().size())
+ assertEquals(topicId, shareAcknowledgeResponseData.responses().get(0).topicId())
+ assertEquals(1, shareAcknowledgeResponseData.responses().get(0).partitions().size())
+
+ val expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+
+ val acknowledgePartitionData = shareAcknowledgeResponseData.responses().get(0).partitions().get(0)
+ compareAcknowledgeResponsePartitions(acknowledgePartitionData, expectedAcknowledgePartitionData)
+
+ // Sending a second share fetch request to check if acknowledgements were done successfully
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(2))) // Records at offsets 0 to 9 should be fetched again because they were released with delivery count as 2
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestPiggybackedRelease(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ var acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Producing 10 more records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ // Send a Share Fetch request with piggybacked acknowledgements
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ acknowledgementsMapForFetch = Map(topicIdPartition -> List(new ShareFetchRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(2.toByte))).asJava) // Release the records
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(List(0L, 10L).asJava, List(9L, 19L).asJava, List(2, 1).asJava))
+ // All the records from offsets 0 to 19 will be fetched. Records from 0 to 9 will have delivery count as 2 because
+ // they are re delivered, and records from 10 to 19 will have delivery count as 1 because they are newly acquired
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareAcknowledgeRequestSuccessReject(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ val acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Send a Share Acknowledge request to acknowledge the fetched records
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ val acknowledgementsMapForAcknowledge: Map[TopicIdPartition, util.List[ShareAcknowledgeRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition -> List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(3.toByte))).asJava) // Reject the records
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMapForAcknowledge)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.NONE.code, shareAcknowledgeResponseData.errorCode)
+ assertEquals(1, shareAcknowledgeResponseData.responses().size())
+ assertEquals(topicId, shareAcknowledgeResponseData.responses().get(0).topicId())
+ assertEquals(1, shareAcknowledgeResponseData.responses().get(0).partitions().size())
+
+ val expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+
+ val acknowledgePartitionData = shareAcknowledgeResponseData.responses().get(0).partitions().get(0)
+ compareAcknowledgeResponsePartitions(acknowledgePartitionData, expectedAcknowledgePartitionData)
+
+ // Producing 10 more records to the topic
+ produceData(topicIdPartition, 10)
+
+ // Sending a second share fetch request to check if acknowledgements were done successfully
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(10), Collections.singletonList(19), Collections.singletonList(1))) // Only the records from offset 10 onwards should be fetched because records at offsets 0-9 have been rejected
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestPiggybackedReject(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ var acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Producing 10 more records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ // Send a Share Fetch request with piggybacked acknowledgements
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ acknowledgementsMapForFetch = Map(topicIdPartition -> List(new ShareFetchRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(3.toByte))).asJava) // Reject the records
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(10), Collections.singletonList(19), Collections.singletonList(1))) // The records at offsets 0 to 9 will not be re fetched because they have been rejected
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Producing 10 more records to the topic
+ produceData(topicIdPartition, 10)
+
+ // Sending a third share fetch request to confirm if acknowledgements were done successfully
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(20), Collections.singletonList(29), Collections.singletonList(1))) // Only the records from offset 20 onwards should be fetched because records at offsets 0-9 have been rejected before and 10 to 19 are currently acquired
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1"),
+ new ClusterConfigProperty(key = "group.share.delivery.count.limit", value = "2") // Setting max delivery count config to 2
+ )
+ )
+ def testShareAcknowledgeRequestMaxDeliveryAttemptExhausted(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ val acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Send a Share Acknowledge request to acknowledge the fetched records
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ var acknowledgementsMapForAcknowledge: Map[TopicIdPartition, util.List[ShareAcknowledgeRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition -> List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(2.toByte))).asJava) // Release the records
+ var shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMapForAcknowledge)
+ var shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ var shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.NONE.code, shareAcknowledgeResponseData.errorCode)
+ assertEquals(1, shareAcknowledgeResponseData.responses().size())
+ assertEquals(topicId, shareAcknowledgeResponseData.responses().get(0).topicId())
+ assertEquals(1, shareAcknowledgeResponseData.responses().get(0).partitions().size())
+
+ var expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+
+ var acknowledgePartitionData = shareAcknowledgeResponseData.responses().get(0).partitions().get(0)
+ compareAcknowledgeResponsePartitions(acknowledgePartitionData, expectedAcknowledgePartitionData)
+
+ // Sending a second share fetch request to check if acknowledgements were done successfully
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(2))) // Records at offsets 0 to 9 should be fetched again because they were released with delivery count as 2
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Send a Share Acknowledge request to acknowledge the fetched records
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ acknowledgementsMapForAcknowledge = Map(topicIdPartition -> List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(2.toByte))).asJava) // Release the records again
+ shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMapForAcknowledge)
+ shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.NONE.code, shareAcknowledgeResponseData.errorCode)
+ assertEquals(1, shareAcknowledgeResponseData.responses().size())
+ assertEquals(topicId, shareAcknowledgeResponseData.responses().get(0).topicId())
+ assertEquals(1, shareAcknowledgeResponseData.responses().get(0).partitions().size())
+
+ expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+
+ acknowledgePartitionData = shareAcknowledgeResponseData.responses().get(0).partitions().get(0)
+ compareAcknowledgeResponsePartitions(acknowledgePartitionData, expectedAcknowledgePartitionData)
+
+ // Producing 10 new records to the topic
+ produceData(topicIdPartition, 10)
+
+ // Sending a third share fetch request to check if acknowledgements were done successfully
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(10), Collections.singletonList(19), Collections.singletonList(1))) // Only new records from offset 10 to 19 will be fetched, records at offsets 0 to 9 have been archived because delivery count limit has been exceeded
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1"),
+ )
+ )
+ def testShareFetchBrokerRespectsPartitionsSizeLimit(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 3 large messages to the topic created above
+ produceData(topicIdPartition, 10)
+ produceData(topicIdPartition, "large message 1", new String(new Array[Byte](MAX_PARTITION_BYTES/3)))
+ produceData(topicIdPartition, "large message 2", new String(new Array[Byte](MAX_PARTITION_BYTES/3)))
+ produceData(topicIdPartition, "large message 3", new String(new Array[Byte](MAX_PARTITION_BYTES/3)))
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ val shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ val expectedPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(11), Collections.singletonList(1)))
+ // The first 10 records will be consumed as it is. For the last 3 records, each of size MAX_PARTITION_BYTES/3,
+ // only 2 of then will be consumed (offsets 10 and 11) because the inclusion of the third last record will exceed
+ // the max partition bytes limit
+
+ val partitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(partitionData, expectedPartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestSuccessfulSharingBetweenMultipleConsumers(): Unit = {
+ val groupId: String = "group"
+ val memberId1 = Uuid.randomUuid()
+ val memberId2 = Uuid.randomUuid()
+ val memberId3 = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10000 records to the topic created above
+ produceData(topicIdPartition, 10000)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Sending 3 share Fetch Requests with same groupId to the same topicPartition but with different memberIds,
+ // mocking the behaviour of multiple share consumers from the same share group
+ val metadata1: ShareRequestMetadata = new ShareRequestMetadata(memberId1, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap1: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest1 = createShareFetchRequest(groupId, metadata1, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap1)
+
+ val metadata2: ShareRequestMetadata = new ShareRequestMetadata(memberId2, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap2: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest2 = createShareFetchRequest(groupId, metadata2, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap2)
+
+ val metadata3: ShareRequestMetadata = new ShareRequestMetadata(memberId3, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap3: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest3 = createShareFetchRequest(groupId, metadata3, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap3)
+
+ val shareFetchResponse1 = connectAndReceive[ShareFetchResponse](shareFetchRequest1)
+ val shareFetchResponse2 = connectAndReceive[ShareFetchResponse](shareFetchRequest2)
+ val shareFetchResponse3 = connectAndReceive[ShareFetchResponse](shareFetchRequest3)
+
+
+ val shareFetchResponseData1 = shareFetchResponse1.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData1.errorCode)
+ assertEquals(1, shareFetchResponseData1.responses().size())
+ assertEquals(topicId, shareFetchResponseData1.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData1.responses().get(0).partitions().size())
+
+ val partitionData1 = shareFetchResponseData1.responses().get(0).partitions().get(0)
+
+ val shareFetchResponseData2 = shareFetchResponse2.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData2.errorCode)
+ assertEquals(1, shareFetchResponseData2.responses().size())
+ assertEquals(topicId, shareFetchResponseData2.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData2.responses().get(0).partitions().size())
+
+ val partitionData2 = shareFetchResponseData2.responses().get(0).partitions().get(0)
+
+ val shareFetchResponseData3 = shareFetchResponse3.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData3.errorCode)
+ assertEquals(1, shareFetchResponseData3.responses().size())
+ assertEquals(topicId, shareFetchResponseData3.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData3.responses().get(0).partitions().size())
+
+ val partitionData3 = shareFetchResponseData3.responses().get(0).partitions().get(0)
+
+ // There should be no common records between the 3 consumers as they are part of the same group
+ assertTrue(partitionData1.acquiredRecords().get(0).lastOffset() < partitionData2.acquiredRecords().get(0).firstOffset())
+ assertTrue(partitionData2.acquiredRecords().get(0).lastOffset() < partitionData3.acquiredRecords().get(0).firstOffset())
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestNoSharingBetweenMultipleConsumersFromDifferentGroups(): Unit = {
+ val groupId1: String = "group1"
+ val groupId2: String = "group2"
+ val groupId3: String = "group3"
+
+ val memberId1 = Uuid.randomUuid()
+ val memberId2 = Uuid.randomUuid()
+ val memberId3 = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Sending 3 share Fetch Requests with same groupId to the same topicPartition but with different memberIds,
+ // mocking the behaviour of multiple share consumers from the same share group
+ val metadata1: ShareRequestMetadata = new ShareRequestMetadata(memberId1, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap1: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest1 = createShareFetchRequest(groupId1, metadata1, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap1)
+
+ val metadata2: ShareRequestMetadata = new ShareRequestMetadata(memberId2, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap2: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest2 = createShareFetchRequest(groupId2, metadata2, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap2)
+
+ val metadata3: ShareRequestMetadata = new ShareRequestMetadata(memberId3, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap3: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ val shareFetchRequest3 = createShareFetchRequest(groupId3, metadata3, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap3)
+
+ val shareFetchResponse1 = connectAndReceive[ShareFetchResponse](shareFetchRequest1)
+ val shareFetchResponse2 = connectAndReceive[ShareFetchResponse](shareFetchRequest2)
+ val shareFetchResponse3 = connectAndReceive[ShareFetchResponse](shareFetchRequest3)
+
+
+ val shareFetchResponseData1 = shareFetchResponse1.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData1.errorCode)
+ assertEquals(1, shareFetchResponseData1.responses().size())
+ assertEquals(topicId, shareFetchResponseData1.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData1.responses().get(0).partitions().size())
+
+ val partitionData1 = shareFetchResponseData1.responses().get(0).partitions().get(0)
+
+ val shareFetchResponseData2 = shareFetchResponse2.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData2.errorCode)
+ assertEquals(1, shareFetchResponseData2.responses().size())
+ assertEquals(topicId, shareFetchResponseData2.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData2.responses().get(0).partitions().size())
+
+ val partitionData2 = shareFetchResponseData2.responses().get(0).partitions().get(0)
+
+ val shareFetchResponseData3 = shareFetchResponse3.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData3.errorCode)
+ assertEquals(1, shareFetchResponseData3.responses().size())
+ assertEquals(topicId, shareFetchResponseData3.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData3.responses().get(0).partitions().size())
+
+ val partitionData3 = shareFetchResponseData3.responses().get(0).partitions().get(0)
+
+ // All the consumers should consume all the records since they are part of different groups
+ assertEquals(partitionData1.acquiredRecords(), expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+ assertEquals(partitionData2.acquiredRecords(), expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+ assertEquals(partitionData3.acquiredRecords(), expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareSessionCloseWithShareFetch(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ var acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Producing 10 more records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ // Send a Share Fetch request with piggybacked acknowledgements
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ acknowledgementsMapForFetch = Map(topicIdPartition -> List(new ShareFetchRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava) // Accept the records
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(10), Collections.singletonList(19), Collections.singletonList(1))) // The records at offsets 0 to 9 will not be re fetched because they have been acknowledged
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Sending a final fetch request to close the session
+ shareSessionEpoch = ShareRequestMetadata.FINAL_EPOCH
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ acknowledgementsMapForFetch = Map(topicIdPartition -> List(new ShareFetchRequestData.AcknowledgementBatch()
+ .setFirstOffset(10)
+ .setLastOffset(19)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava) // Accept the records
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, 0, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(0, shareFetchResponseData.responses().size()) // responses list will be empty because there are no responses for the final fetch request
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareSessionCloseWithShareAcknowledge(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var shareSessionEpoch = ShareRequestMetadata.INITIAL_EPOCH
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ var acknowledgementsMapForFetch: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ var expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ var fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Producing 10 more records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ // Send a Share Fetch request with piggybacked acknowledgements
+ shareSessionEpoch = ShareRequestMetadata.nextEpoch(shareSessionEpoch)
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ acknowledgementsMapForFetch = Map(topicIdPartition -> List(new ShareFetchRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava) // Accept the records
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMapForFetch)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ expectedFetchPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(10), Collections.singletonList(19), Collections.singletonList(1))) // The records at offsets 0 to 9 will not be re fetched because they have been acknowledged
+
+ fetchPartitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(fetchPartitionData, expectedFetchPartitionData)
+
+ // Sending a Share Acknowledge request to close the session
+ shareSessionEpoch = ShareRequestMetadata.FINAL_EPOCH
+ metadata = new ShareRequestMetadata(memberId, shareSessionEpoch)
+ val acknowledgementsMapForAcknowledge: Map[TopicIdPartition, util.List[ShareAcknowledgeRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition -> List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(10)
+ .setLastOffset(19)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava) // Accept the records
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMapForAcknowledge)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.NONE.code, shareAcknowledgeResponseData.errorCode)
+ assertEquals(1, shareAcknowledgeResponseData.responses().size())
+ assertEquals(topicId, shareAcknowledgeResponseData.responses().get(0).topicId())
+ assertEquals(1, shareAcknowledgeResponseData.responses().get(0).partitions().size())
+
+ val expectedAcknowledgePartitionData = new ShareAcknowledgeResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+
+ val acknowledgePartitionData = shareAcknowledgeResponseData.responses().get(0).partitions().get(0)
+ compareAcknowledgeResponsePartitions(acknowledgePartitionData, expectedAcknowledgePartitionData)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchInitialEpochWithAcknowledgements(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition -> List(new ShareFetchRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava) // Acknowledgements in the Initial Fetch Request
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ val shareFetchResponseData = shareFetchResponse.data()
+ // The response will have a top level error code because this is an Initial Fetch request with acknowledgement data present
+ assertEquals(Errors.INVALID_REQUEST.code(), shareFetchResponseData.errorCode)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareAcknowledgeInitialRequestError(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ // Send the share fetch request to fetch the records produced above
+ val metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareAcknowledgeRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition ->
+ List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava)
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMap)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, shareAcknowledgeResponseData.errorCode)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestInvalidShareSessionEpoch(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ val expectedPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val partitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(partitionData, expectedPartitionData)
+
+ // Sending Share Fetch request with invalid share session epoch
+ metadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.nextEpoch(ShareRequestMetadata.nextEpoch(ShareRequestMetadata.INITIAL_EPOCH)))
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, shareFetchResponseData.errorCode)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareAcknowledgeRequestInvalidShareSessionEpoch(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ val shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ val expectedPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val partitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(partitionData, expectedPartitionData)
+
+ // Sending Share Acknowledge request with invalid share session epoch
+ metadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.nextEpoch(ShareRequestMetadata.nextEpoch(ShareRequestMetadata.INITIAL_EPOCH)))
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareAcknowledgeRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition -> List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava)
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMap)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.INVALID_SHARE_SESSION_EPOCH.code, shareAcknowledgeResponseData.errorCode)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestShareSessionNotFound(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+ val wrongMemberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ val expectedPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val partitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(partitionData, expectedPartitionData)
+
+ // Sending a Share Fetch request with wrong member Id
+ metadata = new ShareRequestMetadata(wrongMemberId, ShareRequestMetadata.nextEpoch(ShareRequestMetadata.INITIAL_EPOCH))
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.SHARE_SESSION_NOT_FOUND.code, shareFetchResponseData.errorCode)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareAcknowledgeRequestShareSessionNotFound(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+ val wrongMemberId = Uuid.randomUuid()
+
+ val topic = "topic"
+ val partition = 0
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition = new TopicIdPartition(topicId, new TopicPartition(topic, partition))
+
+ initProducer()
+ // Producing 10 records to the topic created above
+ produceData(topicIdPartition, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition)
+
+ // Send the share fetch request to fetch the records produced above
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, Map.empty)
+ val shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ val shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ val expectedPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(0), Collections.singletonList(9), Collections.singletonList(1)))
+
+ val partitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(partitionData, expectedPartitionData)
+
+ // Sending a Share Acknowledge request with wrong member Id
+ metadata = new ShareRequestMetadata(wrongMemberId, ShareRequestMetadata.nextEpoch(ShareRequestMetadata.INITIAL_EPOCH))
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareAcknowledgeRequestData.AcknowledgementBatch]] =
+ Map(topicIdPartition -> List(new ShareAcknowledgeRequestData.AcknowledgementBatch()
+ .setFirstOffset(0)
+ .setLastOffset(9)
+ .setAcknowledgeTypes(Collections.singletonList(1.toByte))).asJava)
+ val shareAcknowledgeRequest = createShareAcknowledgeRequest(groupId, metadata, acknowledgementsMap)
+ val shareAcknowledgeResponse = connectAndReceive[ShareAcknowledgeResponse](shareAcknowledgeRequest)
+
+ val shareAcknowledgeResponseData = shareAcknowledgeResponse.data()
+ assertEquals(Errors.SHARE_SESSION_NOT_FOUND.code, shareAcknowledgeResponseData.errorCode)
+ }
+
+ @ClusterTest(
+ serverProperties = Array(
+ new ClusterConfigProperty(key = "group.coordinator.rebalance.protocols", value = "classic,consumer,share"),
+ new ClusterConfigProperty(key = "group.share.enable", value = "true"),
+ new ClusterConfigProperty(key = "offsets.topic.num.partitions", value = "1"),
+ new ClusterConfigProperty(key = "offsets.topic.replication.factor", value = "1")
+ )
+ )
+ def testShareFetchRequestForgetTopicPartitions(): Unit = {
+ val groupId: String = "group"
+ val memberId = Uuid.randomUuid()
+
+ val topic = "topic1"
+ val partition1 = 0
+ val partition2 = 1
+
+ createTopicAndReturnLeaders(topic, numPartitions = 3)
+ val topicIds = getTopicIds.asJava
+ val topicId = topicIds.get(topic)
+ val topicIdPartition1 = new TopicIdPartition(topicId, new TopicPartition(topic, partition1))
+ val topicIdPartition2 = new TopicIdPartition(topicId, new TopicPartition(topic, partition2))
+
+ initProducer()
+ // Producing 10 records to the topic partitions created above
+ produceData(topicIdPartition1, 10)
+ produceData(topicIdPartition2, 10)
+
+ val send: Seq[TopicIdPartition] = Seq(topicIdPartition1, topicIdPartition2)
+
+ // Send the share fetch request to fetch the records produced above
+ var metadata: ShareRequestMetadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.INITIAL_EPOCH)
+ val acknowledgementsMap: Map[TopicIdPartition, util.List[ShareFetchRequestData.AcknowledgementBatch]] = Map.empty
+ var shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, send, Seq.empty, acknowledgementsMap)
+ var shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ var shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(2, shareFetchResponseData.responses().get(0).partitions().size())
+
+ // Producing 10 more records to the topic partitions created above
+ produceData(topicIdPartition1, 10)
+ produceData(topicIdPartition2, 10)
+
+ // Send the share fetch request to with forget list populated with topicIdPartition2
+ metadata = new ShareRequestMetadata(memberId, ShareRequestMetadata.nextEpoch(ShareRequestMetadata.INITIAL_EPOCH))
+ val forget: Seq[TopicIdPartition] = Seq(topicIdPartition1)
+ shareFetchRequest = createShareFetchRequest(groupId, metadata, MAX_PARTITION_BYTES, Seq.empty, forget, acknowledgementsMap)
+ shareFetchResponse = connectAndReceive[ShareFetchResponse](shareFetchRequest)
+
+ shareFetchResponseData = shareFetchResponse.data()
+ assertEquals(Errors.NONE.code, shareFetchResponseData.errorCode)
+ assertEquals(1, shareFetchResponseData.responses().size())
+ assertEquals(topicId, shareFetchResponseData.responses().get(0).topicId())
+ assertEquals(1, shareFetchResponseData.responses().get(0).partitions().size())
+
+ val expectedPartitionData = new ShareFetchResponseData.PartitionData()
+ .setPartitionIndex(partition2)
+ .setErrorCode(Errors.NONE.code())
+ .setAcknowledgeErrorCode(Errors.NONE.code())
+ .setAcquiredRecords(expectedAcquiredRecords(Collections.singletonList(10), Collections.singletonList(19), Collections.singletonList(1)))
+
+ val partitionData = shareFetchResponseData.responses().get(0).partitions().get(0)
+ compareFetchResponsePartitions(partitionData, expectedPartitionData)
+ }
+
+ private def expectedAcquiredRecords(firstOffsets: util.List[Long], lastOffsets: util.List[Long], deliveryCounts: util.List[Int]): util.List[AcquiredRecords] = {
+ val acquiredRecordsList: util.List[AcquiredRecords] = new util.ArrayList()
+ for (i <- firstOffsets.indices) {
+ acquiredRecordsList.add(new AcquiredRecords()
+ .setFirstOffset(firstOffsets.get(i))
+ .setLastOffset(lastOffsets.get(i))
+ .setDeliveryCount(deliveryCounts.get(i).toShort))
+ }
+ acquiredRecordsList
+ }
+
+ private def compareFetchResponsePartitions(actualResponse: ShareFetchResponseData.PartitionData,
+ expectedResponse: ShareFetchResponseData.PartitionData): Unit = {
+ assertEquals(actualResponse.partitionIndex(), expectedResponse.partitionIndex())
Review Comment:
nit: expected should be first parameter and then actual
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1723903760
##########
core/src/test/scala/unit/kafka/server/RequestUtilitiesTest.scala:
##########
@@ -0,0 +1,152 @@
+/**
+ * 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 unit.kafka.server
+
+import kafka.network.SocketServer
+import kafka.server.{ControllerServer, IntegrationTestUtils, KafkaBroker}
+import kafka.test.ClusterInstance
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse}
+import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.serialization.StringSerializer
+import org.apache.kafka.controller.ControllerRequestContextUtil.ANONYMOUS_CONTEXT
+import org.junit.jupiter.api.AfterEach
+
+import java.util.Properties
+import java.util.stream.Collectors
+import scala.collection.Seq
+import scala.collection.convert.ImplicitConversions.{`collection AsScalaIterable`, `map AsScala`}
+import scala.reflect.ClassTag
+
+class RequestUtilitiesTest(cluster: ClusterInstance) {
+
+ protected var producer: KafkaProducer[String, String] = _
+
+ protected def securityProtocol: SecurityProtocol = SecurityProtocol.PLAINTEXT
+ protected def listenerName: ListenerName = ListenerName.forSecurityProtocol(securityProtocol)
+
+ private def brokers(): Seq[KafkaBroker] = cluster.brokers.values().stream().collect(Collectors.toList[KafkaBroker]).toSeq
+
+ private def controllerServers(): Seq[ControllerServer] = cluster.controllers().values().toSeq
+
+ @AfterEach
+ def tearDown(): Unit = {
+ if (producer != null)
+ producer.close()
+ }
+
+ protected def createOffsetsTopic(): Unit = {
+ TestUtils.createOffsetsTopicWithAdmin(
+ admin = cluster.createAdminClient(),
+ brokers = brokers(),
+ controllers = controllerServers()
+ )
+ }
+
+ protected def createTopic(topic: String,
Review Comment:
This is good and potentially we should re-use this topic at other places in the test files. It will be good to create a jira atleast, outside kip-932, for the cleanup.
##########
core/src/test/scala/unit/kafka/server/RequestUtilitiesTest.scala:
##########
@@ -0,0 +1,152 @@
+/**
+ * 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 unit.kafka.server
+
+import kafka.network.SocketServer
+import kafka.server.{ControllerServer, IntegrationTestUtils, KafkaBroker}
+import kafka.test.ClusterInstance
+import kafka.utils.TestUtils
+import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord, RecordMetadata}
+import org.apache.kafka.common.{TopicIdPartition, TopicPartition, Uuid}
+import org.apache.kafka.common.network.ListenerName
+import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse}
+import org.apache.kafka.common.security.auth.SecurityProtocol
+import org.apache.kafka.common.serialization.StringSerializer
+import org.apache.kafka.controller.ControllerRequestContextUtil.ANONYMOUS_CONTEXT
+import org.junit.jupiter.api.AfterEach
+
+import java.util.Properties
+import java.util.stream.Collectors
+import scala.collection.Seq
+import scala.collection.convert.ImplicitConversions.{`collection AsScalaIterable`, `map AsScala`}
+import scala.reflect.ClassTag
+
+class RequestUtilitiesTest(cluster: ClusterInstance) {
Review Comment:
Is it a utility class or a test class? Seems to be utility class so should it be suffixed with `Test` or just `RequestUtilities` should be the right name?
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -25,48 +25,14 @@ import org.apache.kafka.common.message.LeaveGroupResponseData.MemberResponse
import org.apache.kafka.common.message.SyncGroupRequestData.SyncGroupRequestAssignment
import org.apache.kafka.common.message.{ConsumerGroupDescribeRequestData, ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsRequestData, DeleteGroupsResponseData, DescribeGroupsRequestData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, OffsetFetchResponseData, ShareGroupDescribeRequestData, ShareGroupDescribeResponseData, ShareGroupHeartbeatRequestData, ShareGroupHeartbeatResponseData, SyncGroupRequestData, SyncGroupResponseData}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse, DescribeGroupsRequest, DescribeGroupsResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsRequest, ListGroupsResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetDeleteRequest, OffsetDeleteResponse, OffsetFetchRequest, OffsetFetchResponse, ShareGroupDescribeRequest, ShareGroupDescribeResponse, ShareGroupHeartbeatRequest, ShareGroupHeartbeatResponse, SyncGroupRequest, SyncGroupResponse}
+import org.apache.kafka.common.requests.{ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse, DescribeGroupsRequest, DescribeGroupsResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsRequest, ListGroupsResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetDeleteRequest, OffsetDeleteResponse, OffsetFetchRequest, OffsetFetchResponse, ShareGroupDescribeRequest, ShareGroupDescribeResponse, ShareGroupHeartbeatRequest, ShareGroupHeartbeatResponse, SyncGroupRequest, SyncGroupResponse}
import org.junit.jupiter.api.Assertions.{assertEquals, fail}
+import unit.kafka.server.RequestUtilitiesTest
import java.util.Comparator
-import java.util.stream.Collectors
import scala.jdk.CollectionConverters._
-import scala.reflect.ClassTag
-class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
- private def brokers(): Seq[KafkaBroker] = cluster.brokers.values().stream().collect(Collectors.toList[KafkaBroker]).asScala.toSeq
-
- private def controllerServers(): Seq[ControllerServer] = cluster.controllers().values().asScala.toSeq
-
- protected def createOffsetsTopic(): Unit = {
- TestUtils.createOffsetsTopicWithAdmin(
- admin = cluster.createAdminClient(),
- brokers = brokers(),
- controllers = controllerServers()
- )
- }
-
- protected def createTopic(
- topic: String,
- numPartitions: Int
- ): Unit = {
- TestUtils.createTopicWithAdmin(
- admin = cluster.createAdminClient(),
- brokers = brokers(),
- controllers = controllerServers(),
- topic = topic,
- numPartitions = numPartitions
- )
- }
-
- protected def isUnstableApiEnabled: Boolean = {
- cluster.config.serverProperties.get("unstable.api.versions.enable") == "true"
- }
-
- protected def isNewGroupCoordinatorEnabled: Boolean = {
- cluster.config.serverProperties.get("group.coordinator.new.enable") == "true" ||
- cluster.config.serverProperties.get("group.coordinator.rebalance.protocols").contains("consumer")
- }
+class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) extends RequestUtilitiesTest(cluster) {
Review Comment:
Why the test classes should extend a utility class, seems not right.
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1731724112
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -25,48 +25,14 @@ import org.apache.kafka.common.message.LeaveGroupResponseData.MemberResponse
import org.apache.kafka.common.message.SyncGroupRequestData.SyncGroupRequestAssignment
import org.apache.kafka.common.message.{ConsumerGroupDescribeRequestData, ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsRequestData, DeleteGroupsResponseData, DescribeGroupsRequestData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, OffsetFetchResponseData, ShareGroupDescribeRequestData, ShareGroupDescribeResponseData, ShareGroupHeartbeatRequestData, ShareGroupHeartbeatResponseData, SyncGroupRequestData, SyncGroupResponseData}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse, DescribeGroupsRequest, DescribeGroupsResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsRequest, ListGroupsResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetDeleteRequest, OffsetDeleteResponse, OffsetFetchRequest, OffsetFetchResponse, ShareGroupDescribeRequest, ShareGroupDescribeResponse, ShareGroupHeartbeatRequest, ShareGroupHeartbeatResponse, SyncGroupRequest, SyncGroupResponse}
+import org.apache.kafka.common.requests.{ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse, DescribeGroupsRequest, DescribeGroupsResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsRequest, ListGroupsResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetDeleteRequest, OffsetDeleteResponse, OffsetFetchRequest, OffsetFetchResponse, ShareGroupDescribeRequest, ShareGroupDescribeResponse, ShareGroupHeartbeatRequest, ShareGroupHeartbeatResponse, SyncGroupRequest, SyncGroupResponse}
import org.junit.jupiter.api.Assertions.{assertEquals, fail}
+import unit.kafka.server.RequestUtilitiesTest
import java.util.Comparator
-import java.util.stream.Collectors
import scala.jdk.CollectionConverters._
-import scala.reflect.ClassTag
-class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
- private def brokers(): Seq[KafkaBroker] = cluster.brokers.values().stream().collect(Collectors.toList[KafkaBroker]).asScala.toSeq
-
- private def controllerServers(): Seq[ControllerServer] = cluster.controllers().values().asScala.toSeq
-
- protected def createOffsetsTopic(): Unit = {
- TestUtils.createOffsetsTopicWithAdmin(
- admin = cluster.createAdminClient(),
- brokers = brokers(),
- controllers = controllerServers()
- )
- }
-
- protected def createTopic(
- topic: String,
- numPartitions: Int
- ): Unit = {
- TestUtils.createTopicWithAdmin(
- admin = cluster.createAdminClient(),
- brokers = brokers(),
- controllers = controllerServers(),
- topic = topic,
- numPartitions = numPartitions
- )
- }
-
- protected def isUnstableApiEnabled: Boolean = {
- cluster.config.serverProperties.get("unstable.api.versions.enable") == "true"
- }
-
- protected def isNewGroupCoordinatorEnabled: Boolean = {
- cluster.config.serverProperties.get("group.coordinator.new.enable") == "true" ||
- cluster.config.serverProperties.get("group.coordinator.rebalance.protocols").contains("consumer")
- }
+class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) extends RequestUtilitiesTest(cluster) {
Review Comment:
Thanks for the review. Yep that seems better. I have pushed in the new changes in the latest commit
--
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-17290: Added Integration tests for ShareFetch and ShareAcknowledge APIs [kafka]
Posted by "chirag-wadhwa5 (via GitHub)" <gi...@apache.org>.
chirag-wadhwa5 commented on code in PR #16916:
URL: https://github.com/apache/kafka/pull/16916#discussion_r1724504350
##########
core/src/test/scala/unit/kafka/server/GroupCoordinatorBaseRequestTest.scala:
##########
@@ -25,48 +25,14 @@ import org.apache.kafka.common.message.LeaveGroupResponseData.MemberResponse
import org.apache.kafka.common.message.SyncGroupRequestData.SyncGroupRequestAssignment
import org.apache.kafka.common.message.{ConsumerGroupDescribeRequestData, ConsumerGroupDescribeResponseData, ConsumerGroupHeartbeatRequestData, ConsumerGroupHeartbeatResponseData, DeleteGroupsRequestData, DeleteGroupsResponseData, DescribeGroupsRequestData, DescribeGroupsResponseData, HeartbeatRequestData, HeartbeatResponseData, JoinGroupRequestData, JoinGroupResponseData, LeaveGroupResponseData, ListGroupsRequestData, ListGroupsResponseData, OffsetCommitRequestData, OffsetCommitResponseData, OffsetDeleteRequestData, OffsetDeleteResponseData, OffsetFetchResponseData, ShareGroupDescribeRequestData, ShareGroupDescribeResponseData, ShareGroupHeartbeatRequestData, ShareGroupHeartbeatResponseData, SyncGroupRequestData, SyncGroupResponseData}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
-import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse, DescribeGroupsRequest, DescribeGroupsResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsRequest, ListGroupsResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetDeleteRequest, OffsetDeleteResponse, OffsetFetchRequest, OffsetFetchResponse, ShareGroupDescribeRequest, ShareGroupDescribeResponse, ShareGroupHeartbeatRequest, ShareGroupHeartbeatResponse, SyncGroupRequest, SyncGroupResponse}
+import org.apache.kafka.common.requests.{ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerGroupHeartbeatRequest, ConsumerGroupHeartbeatResponse, DeleteGroupsRequest, DeleteGroupsResponse, DescribeGroupsRequest, DescribeGroupsResponse, HeartbeatRequest, HeartbeatResponse, JoinGroupRequest, JoinGroupResponse, LeaveGroupRequest, LeaveGroupResponse, ListGroupsRequest, ListGroupsResponse, OffsetCommitRequest, OffsetCommitResponse, OffsetDeleteRequest, OffsetDeleteResponse, OffsetFetchRequest, OffsetFetchResponse, ShareGroupDescribeRequest, ShareGroupDescribeResponse, ShareGroupHeartbeatRequest, ShareGroupHeartbeatResponse, SyncGroupRequest, SyncGroupResponse}
import org.junit.jupiter.api.Assertions.{assertEquals, fail}
+import unit.kafka.server.RequestUtilitiesTest
import java.util.Comparator
-import java.util.stream.Collectors
import scala.jdk.CollectionConverters._
-import scala.reflect.ClassTag
-class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) {
- private def brokers(): Seq[KafkaBroker] = cluster.brokers.values().stream().collect(Collectors.toList[KafkaBroker]).asScala.toSeq
-
- private def controllerServers(): Seq[ControllerServer] = cluster.controllers().values().asScala.toSeq
-
- protected def createOffsetsTopic(): Unit = {
- TestUtils.createOffsetsTopicWithAdmin(
- admin = cluster.createAdminClient(),
- brokers = brokers(),
- controllers = controllerServers()
- )
- }
-
- protected def createTopic(
- topic: String,
- numPartitions: Int
- ): Unit = {
- TestUtils.createTopicWithAdmin(
- admin = cluster.createAdminClient(),
- brokers = brokers(),
- controllers = controllerServers(),
- topic = topic,
- numPartitions = numPartitions
- )
- }
-
- protected def isUnstableApiEnabled: Boolean = {
- cluster.config.serverProperties.get("unstable.api.versions.enable") == "true"
- }
-
- protected def isNewGroupCoordinatorEnabled: Boolean = {
- cluster.config.serverProperties.get("group.coordinator.new.enable") == "true" ||
- cluster.config.serverProperties.get("group.coordinator.rebalance.protocols").contains("consumer")
- }
+class GroupCoordinatorBaseRequestTest(cluster: ClusterInstance) extends RequestUtilitiesTest(cluster) {
Review Comment:
Thanks for the review. It is a utilities class, but as per the implementation of the KafkaClusterTestKit, the cluster created as part of the test is passed on to the parent class via the primary constructor.
```class RequestUtilitiesTest(cluster: ClusterInstance) {```
And we need this cluster instance for the other utility methods to work, that's why it is being inherited.
--
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