You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/11/22 10:12:29 UTC
[GitHub] [kafka] showuon commented on a diff in pull request #12845: KAFKA-14367; Add `JoinGroup` to the new `GroupCoordinator` interface
showuon commented on code in PR #12845:
URL: https://github.com/apache/kafka/pull/12845#discussion_r1029114602
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorRequestContext.java:
##########
@@ -0,0 +1,74 @@
+/*
+ * 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 org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.utils.BufferSupplier;
+
+import java.net.InetAddress;
+import java.util.Objects;
+
+public class GroupCoordinatorRequestContext {
Review Comment:
nit: Should we add `toString` for debugging?
##########
core/src/test/scala/unit/kafka/server/KafkaApisTest.scala:
##########
@@ -2524,196 +2528,207 @@ class KafkaApisTest {
assertEquals(MemoryRecords.EMPTY, FetchResponse.recordsOrFail(partitionData))
}
- @Test
- def testJoinGroupProtocolsOrder(): Unit = {
- val protocols = List(
- ("first", "first".getBytes()),
- ("second", "second".getBytes())
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.JOIN_GROUP)
+ def testHandleJoinGroupRequest(version: Short): Unit = {
+ val joinGroupRequest = new JoinGroupRequestData()
+ .setGroupId("group")
+ .setMemberId("member")
+ .setProtocolType("consumer")
+ .setRebalanceTimeoutMs(1000)
+ .setSessionTimeoutMs(2000)
+
+ val requestChannelRequest = buildRequest(new JoinGroupRequest.Builder(joinGroupRequest).build(version))
+
+ val expectedRequestContext = new GroupCoordinatorRequestContext(
+ version,
+ requestChannelRequest.context.clientId,
+ requestChannelRequest.context.clientAddress,
+ RequestLocal.NoCaching.bufferSupplier
)
- val groupId = "group"
- val memberId = "member1"
- val protocolType = "consumer"
- val rebalanceTimeoutMs = 10
- val sessionTimeoutMs = 5
- val capturedProtocols: ArgumentCaptor[List[(String, Array[Byte])]] = ArgumentCaptor.forClass(classOf[List[(String, Array[Byte])]])
+ val expectedJoinGroupRequest = new JoinGroupRequestData()
+ .setGroupId(joinGroupRequest.groupId)
+ .setMemberId(joinGroupRequest.memberId)
+ .setProtocolType(joinGroupRequest.protocolType)
+ .setRebalanceTimeoutMs(if (version >= 1) joinGroupRequest.rebalanceTimeoutMs else joinGroupRequest.sessionTimeoutMs)
+ .setSessionTimeoutMs(joinGroupRequest.sessionTimeoutMs)
- createKafkaApis().handleJoinGroupRequest(
- buildRequest(
- new JoinGroupRequest.Builder(
- new JoinGroupRequestData()
- .setGroupId(groupId)
- .setMemberId(memberId)
- .setProtocolType(protocolType)
- .setRebalanceTimeoutMs(rebalanceTimeoutMs)
- .setSessionTimeoutMs(sessionTimeoutMs)
- .setProtocols(new JoinGroupRequestData.JoinGroupRequestProtocolCollection(
- protocols.map { case (name, protocol) => new JoinGroupRequestProtocol()
- .setName(name).setMetadata(protocol)
- }.iterator.asJava))
- ).build()
- ),
- RequestLocal.withThreadConfinedCaching)
+ val future = new CompletableFuture[JoinGroupResponseData]()
+ when(newGroupCoordinator.joinGroup(
+ ArgumentMatchers.eq(expectedRequestContext),
+ ArgumentMatchers.eq(expectedJoinGroupRequest)
+ )).thenReturn(future)
- verify(groupCoordinator).handleJoinGroup(
- ArgumentMatchers.eq(groupId),
- ArgumentMatchers.eq(memberId),
- ArgumentMatchers.eq(None),
- ArgumentMatchers.eq(true),
- ArgumentMatchers.eq(true),
- ArgumentMatchers.eq(clientId),
- ArgumentMatchers.eq(InetAddress.getLocalHost.toString),
- ArgumentMatchers.eq(rebalanceTimeoutMs),
- ArgumentMatchers.eq(sessionTimeoutMs),
- ArgumentMatchers.eq(protocolType),
- capturedProtocols.capture(),
- any(),
- any(),
- any()
+ createKafkaApis().handleJoinGroupRequest(
+ requestChannelRequest,
+ RequestLocal.NoCaching
)
- val capturedProtocolsList = capturedProtocols.getValue
- assertEquals(protocols.size, capturedProtocolsList.size)
- protocols.zip(capturedProtocolsList).foreach { case ((expectedName, expectedBytes), (name, bytes)) =>
- assertEquals(expectedName, name)
- assertArrayEquals(expectedBytes, bytes)
- }
- }
- @Test
- def testJoinGroupWhenAnErrorOccurs(): Unit = {
- for (version <- ApiKeys.JOIN_GROUP.oldestVersion to ApiKeys.JOIN_GROUP.latestVersion) {
- testJoinGroupWhenAnErrorOccurs(version.asInstanceOf[Short])
- }
- }
+ val expectedJoinGroupResponse = new JoinGroupResponseData()
+ .setMemberId("member")
+ .setGenerationId(0)
+ .setLeader("leader")
+ .setProtocolType("consumer")
+ .setProtocolName("range")
- def testJoinGroupWhenAnErrorOccurs(version: Short): Unit = {
- reset(groupCoordinator, clientRequestQuotaManager, requestChannel, replicaManager)
+ future.complete(expectedJoinGroupResponse)
+ val capturedResponse = verifyNoThrottling(requestChannelRequest)
+ val response = capturedResponse.getValue.asInstanceOf[JoinGroupResponse]
+ assertEquals(expectedJoinGroupResponse, response.data)
+ }
- val groupId = "group"
- val memberId = "member1"
- val protocolType = "consumer"
- val rebalanceTimeoutMs = 10
- val sessionTimeoutMs = 5
+ @ParameterizedTest
+ @ApiKeyVersionsSource(apiKey = ApiKeys.JOIN_GROUP)
+ def testJoinGroupProtocolNameBackwardCompatibility(version: Short): Unit = {
+ val joinGroupRequest = new JoinGroupRequestData()
+ .setGroupId("group")
+ .setMemberId("member")
+ .setProtocolType("consumer")
+ .setRebalanceTimeoutMs(1000)
+ .setSessionTimeoutMs(2000)
+
+ val requestChannelRequest = buildRequest(new JoinGroupRequest.Builder(joinGroupRequest).build(version))
+
+ val expectedRequestContext = new GroupCoordinatorRequestContext(
+ version,
+ requestChannelRequest.context.clientId,
+ requestChannelRequest.context.clientAddress,
+ RequestLocal.NoCaching.bufferSupplier
+ )
- val capturedCallback: ArgumentCaptor[JoinGroupCallback] = ArgumentCaptor.forClass(classOf[JoinGroupCallback])
+ val expectedJoinGroupRequest = new JoinGroupRequestData()
+ .setGroupId(joinGroupRequest.groupId)
+ .setMemberId(joinGroupRequest.memberId)
+ .setProtocolType(joinGroupRequest.protocolType)
+ .setRebalanceTimeoutMs(if (version >= 1) joinGroupRequest.rebalanceTimeoutMs else joinGroupRequest.sessionTimeoutMs)
+ .setSessionTimeoutMs(joinGroupRequest.sessionTimeoutMs)
- val joinGroupRequest = new JoinGroupRequest.Builder(
- new JoinGroupRequestData()
- .setGroupId(groupId)
- .setMemberId(memberId)
- .setProtocolType(protocolType)
- .setRebalanceTimeoutMs(rebalanceTimeoutMs)
- .setSessionTimeoutMs(sessionTimeoutMs)
- ).build(version)
+ val future = new CompletableFuture[JoinGroupResponseData]()
+ when(newGroupCoordinator.joinGroup(
+ ArgumentMatchers.eq(expectedRequestContext),
+ ArgumentMatchers.eq(expectedJoinGroupRequest)
+ )).thenReturn(future)
- val requestChannelRequest = buildRequest(joinGroupRequest)
+ createKafkaApis().handleJoinGroupRequest(
+ requestChannelRequest,
+ RequestLocal.NoCaching
+ )
- createKafkaApis().handleJoinGroupRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
+ val joinGroupResponse = new JoinGroupResponseData()
+ .setErrorCode(Errors.INCONSISTENT_GROUP_PROTOCOL.code)
+ .setMemberId("member")
+ .setProtocolName(null)
- verify(groupCoordinator).handleJoinGroup(
- ArgumentMatchers.eq(groupId),
- ArgumentMatchers.eq(memberId),
- ArgumentMatchers.eq(None),
- ArgumentMatchers.eq(if (version >= 4) true else false),
- ArgumentMatchers.eq(if (version >= 9) true else false),
- ArgumentMatchers.eq(clientId),
- ArgumentMatchers.eq(InetAddress.getLocalHost.toString),
- ArgumentMatchers.eq(if (version >= 1) rebalanceTimeoutMs else sessionTimeoutMs),
- ArgumentMatchers.eq(sessionTimeoutMs),
- ArgumentMatchers.eq(protocolType),
- ArgumentMatchers.eq(List.empty),
- capturedCallback.capture(),
- any(),
- any()
- )
- capturedCallback.getValue.apply(JoinGroupResult(memberId, Errors.INCONSISTENT_GROUP_PROTOCOL))
+ val expectedJoinGroupResponse = new JoinGroupResponseData()
+ .setErrorCode(Errors.INCONSISTENT_GROUP_PROTOCOL.code)
+ .setMemberId("member")
+ .setProtocolName(if (version >= 7) null else GroupCoordinator.NoProtocol)
+ future.complete(joinGroupResponse)
val capturedResponse = verifyNoThrottling(requestChannelRequest)
val response = capturedResponse.getValue.asInstanceOf[JoinGroupResponse]
-
- assertEquals(Errors.INCONSISTENT_GROUP_PROTOCOL, response.error)
- assertEquals(0, response.data.members.size)
- assertEquals(memberId, response.data.memberId)
- assertEquals(GroupCoordinator.NoGeneration, response.data.generationId)
- assertEquals(GroupCoordinator.NoLeader, response.data.leader)
- assertNull(response.data.protocolType)
-
- if (version >= 7) {
- assertNull(response.data.protocolName)
- } else {
- assertEquals(GroupCoordinator.NoProtocol, response.data.protocolName)
- }
+ assertEquals(expectedJoinGroupResponse, response.data)
}
@Test
- def testJoinGroupProtocolType(): Unit = {
- for (version <- ApiKeys.JOIN_GROUP.oldestVersion to ApiKeys.JOIN_GROUP.latestVersion) {
- testJoinGroupProtocolType(version.asInstanceOf[Short])
- }
- }
+ def testHandleJoinGroupRequestFutureFailed(): Unit = {
+ val joinGroupRequest = new JoinGroupRequestData()
+ .setGroupId("group")
+ .setMemberId("member")
+ .setProtocolType("consumer")
+ .setRebalanceTimeoutMs(1000)
+ .setSessionTimeoutMs(2000)
- def testJoinGroupProtocolType(version: Short): Unit = {
- reset(groupCoordinator, clientRequestQuotaManager, requestChannel, replicaManager)
+ val requestChannelRequest = buildRequest(new JoinGroupRequest.Builder(joinGroupRequest).build())
- val groupId = "group"
- val memberId = "member1"
- val protocolType = "consumer"
- val protocolName = "range"
- val rebalanceTimeoutMs = 10
- val sessionTimeoutMs = 5
+ val expectedRequestContext = new GroupCoordinatorRequestContext(
+ ApiKeys.JOIN_GROUP.latestVersion,
+ requestChannelRequest.context.clientId,
+ requestChannelRequest.context.clientAddress,
+ RequestLocal.NoCaching.bufferSupplier
+ )
- val capturedCallback: ArgumentCaptor[JoinGroupCallback] = ArgumentCaptor.forClass(classOf[JoinGroupCallback])
+ val future = new CompletableFuture[JoinGroupResponseData]()
+ when(newGroupCoordinator.joinGroup(
+ ArgumentMatchers.eq(expectedRequestContext),
+ ArgumentMatchers.eq(joinGroupRequest)
+ )).thenReturn(future)
- val joinGroupRequest = new JoinGroupRequest.Builder(
- new JoinGroupRequestData()
- .setGroupId(groupId)
- .setMemberId(memberId)
- .setProtocolType(protocolType)
- .setRebalanceTimeoutMs(rebalanceTimeoutMs)
- .setSessionTimeoutMs(sessionTimeoutMs)
- ).build(version)
+ createKafkaApis().handleJoinGroupRequest(
+ requestChannelRequest,
+ RequestLocal.NoCaching
+ )
- val requestChannelRequest = buildRequest(joinGroupRequest)
+ future.completeExceptionally(Errors.REQUEST_TIMED_OUT.exception)
+ val capturedResponse = verifyNoThrottling(requestChannelRequest)
+ val response = capturedResponse.getValue.asInstanceOf[JoinGroupResponse]
+ assertEquals(Errors.REQUEST_TIMED_OUT, response.error)
+ }
- createKafkaApis().handleJoinGroupRequest(requestChannelRequest, RequestLocal.withThreadConfinedCaching)
+ @Test
+ def testHandleJoinGroupRequestAuthorizationFailed(): Unit = {
+ val joinGroupRequest = new JoinGroupRequestData()
+ .setGroupId("group")
+ .setMemberId("member")
+ .setProtocolType("consumer")
+ .setRebalanceTimeoutMs(1000)
+ .setSessionTimeoutMs(2000)
- verify(groupCoordinator).handleJoinGroup(
- ArgumentMatchers.eq(groupId),
- ArgumentMatchers.eq(memberId),
- ArgumentMatchers.eq(None),
- ArgumentMatchers.eq(if (version >= 4) true else false),
- ArgumentMatchers.eq(if (version >= 9) true else false),
- ArgumentMatchers.eq(clientId),
- ArgumentMatchers.eq(InetAddress.getLocalHost.toString),
- ArgumentMatchers.eq(if (version >= 1) rebalanceTimeoutMs else sessionTimeoutMs),
- ArgumentMatchers.eq(sessionTimeoutMs),
- ArgumentMatchers.eq(protocolType),
- ArgumentMatchers.eq(List.empty),
- capturedCallback.capture(),
- any(),
- any()
+ val requestChannelRequest = buildRequest(new JoinGroupRequest.Builder(joinGroupRequest).build())
+
+ val authorizer: Authorizer = mock(classOf[Authorizer])
+ when(authorizer.authorize(any[RequestContext], any[util.List[Action]]))
+ .thenReturn(Seq(AuthorizationResult.DENIED).asJava)
+
+ createKafkaApis(authorizer = Some(authorizer)).handleJoinGroupRequest(
+ requestChannelRequest,
+ RequestLocal.NoCaching
)
- capturedCallback.getValue.apply(JoinGroupResult(
- members = List.empty,
- memberId = memberId,
- generationId = 0,
- protocolType = Some(protocolType),
- protocolName = Some(protocolName),
- leaderId = memberId,
- skipAssignment = true,
- error = Errors.NONE
- ))
+
val capturedResponse = verifyNoThrottling(requestChannelRequest)
val response = capturedResponse.getValue.asInstanceOf[JoinGroupResponse]
+ assertEquals(Errors.GROUP_AUTHORIZATION_FAILED, response.error)
+ }
- assertEquals(Errors.NONE, response.error)
- assertEquals(0, response.data.members.size)
- assertEquals(memberId, response.data.memberId)
- assertEquals(0, response.data.generationId)
- assertEquals(memberId, response.data.leader)
- assertEquals(protocolName, response.data.protocolName)
- assertEquals(protocolType, response.data.protocolType)
- assertTrue(response.data.skipAssignment)
+ @Test
+ def testHandleJoinGroupRequestUnexpectedException(): Unit = {
+ val joinGroupRequest = new JoinGroupRequestData()
+ .setGroupId("group")
+ .setMemberId("member")
+ .setProtocolType("consumer")
+ .setRebalanceTimeoutMs(1000)
+ .setSessionTimeoutMs(2000)
+
+ val requestChannelRequest = buildRequest(new JoinGroupRequest.Builder(joinGroupRequest).build())
+
+ val expectedRequestContext = new GroupCoordinatorRequestContext(
+ ApiKeys.JOIN_GROUP.latestVersion,
+ requestChannelRequest.context.clientId,
+ requestChannelRequest.context.clientAddress,
+ RequestLocal.NoCaching.bufferSupplier
+ )
+
+ val future = new CompletableFuture[JoinGroupResponseData]()
+ when(newGroupCoordinator.joinGroup(
+ ArgumentMatchers.eq(expectedRequestContext),
+ ArgumentMatchers.eq(joinGroupRequest)
+ )).thenReturn(future)
+
+ var response: JoinGroupResponse = null
+ when(requestChannel.sendResponse(any(), any(), any())).thenAnswer { _ =>
+ throw new Exception("Something went wrong")
+ }.thenAnswer { invocation =>
+ response = invocation.getArgument(1, classOf[JoinGroupResponse])
+ }
+
+ createKafkaApis().handle(
+ requestChannelRequest,
+ RequestLocal.NoCaching
+ )
+
+ future.completeExceptionally(Errors.NOT_COORDINATOR.exception)
+ assertEquals(Errors.UNKNOWN_SERVER_ERROR, response.error)
Review Comment:
nit: Could we add a comment here to mention we expected to catch the exception in `sendResponse`, and handleError will handle it? Otherwise, it is confusing why don't we get NOT_COORDINATOR error in response.
--
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