Skip to content

KAFKA-9844; Maximum number of members within a group is not always enforced due to a race condition in join group #8454

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hachikuji merged 5 commits into from
Apr 23, 2020
Merged

KAFKA-9844; Maximum number of members within a group is not always enforced due to a race condition in join group #8454

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
22e7349
KAFKA-9844; Maximum number of members within a group is not always en…
dajac Apr 9, 2020
1dab5f1
fixup
dajac Apr 9, 2020
16bb5e4
add unit test
dajac Apr 13, 2020
dce9dd2
fixup
dajac Apr 14, 2020
c08ba95
Address review
dajac Apr 17, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 5 additions & 11 deletions core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,17 +145,11 @@ class GroupCoordinator(val brokerId: Int,
responseCallback(JoinGroupResult(memberId, Errors.INVALID_SESSION_TIMEOUT))
} else {
val isUnknownMember = memberId == JoinGroupRequest.UNKNOWN_MEMBER_ID
groupManager.getGroup(groupId) match {
// group is created if it does not exist and the member id is UNKNOWN. if member
// is specified but group does not exist, request is rejected with UNKNOWN_MEMBER_ID
groupManager.getGroup(groupId, isUnknownMember) match {
case None =>
// only try to create the group if the group is UNKNOWN AND
// the member id is UNKNOWN, if member is specified but group does not
// exist we should reject the request.
if (isUnknownMember) {
val group = groupManager.addGroup(new GroupMetadata(groupId, Empty, time))
doUnknownJoinGroup(group, groupInstanceId, requireKnownMemberId, clientId, clientHost, rebalanceTimeoutMs, sessionTimeoutMs, protocolType, protocols, responseCallback)
} else {
responseCallback(JoinGroupResult(memberId, Errors.UNKNOWN_MEMBER_ID))
}
responseCallback(JoinGroupResult(memberId, Errors.UNKNOWN_MEMBER_ID))
case Some(group) =>
group.inLock {
if ((groupIsOverCapacity(group)
Expand All @@ -175,9 +169,9 @@ class GroupCoordinator(val brokerId: Int,
joinPurgatory.checkAndComplete(GroupKey(group.groupId))
}
}
}
}
}
}

private def doUnknownJoinGroup(group: GroupMetadata,
groupInstanceId: Option[String],
Expand Down
10 changes: 7 additions & 3 deletions core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -209,10 +209,14 @@ class GroupMetadataManager(brokerId: Int,
false
}
/**
* Get the group associated with the given groupId, or null if not found
* Get the group associated with the given groupId - the group is created if createIfNotExist
* is true - or null if not found
*/
def getGroup(groupId: String): Option[GroupMetadata] = {
Option(groupMetadataCache.get(groupId))
def getGroup(groupId: String, createIfNotExist: Boolean = false): Option[GroupMetadata] = {

@hachikuji hachikuji Apr 16, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Optional arguments are best avoided. Maybe we could split this into two methods getOrCreateGroup and getGroup?

@dajac dajac Apr 17, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure. I went with the following method getOrMaybeCreateGroup which still takes the createIfNotExist argument in order to keep the calling side concise. It avoids having to switch between the two on the calling side which makes the error handling a little easier to follow. This is perhaps what you had in mind already.

if (createIfNotExist)
Option(groupMetadataCache.getAndMaybePut(groupId, new GroupMetadata(groupId, Empty, time)))
else
Option(groupMetadataCache.get(groupId))
}

/**
Expand Down
35 changes: 33 additions & 2 deletions core/src/test/scala/unit/kafka/coordinator/group/GroupCoordinatorConcurrencyTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

package kafka.coordinator.group

import java.util.Properties
import java.util.concurrent.{ConcurrentHashMap, TimeUnit}

import kafka.common.OffsetAndMetadata
Expand Down Expand Up @@ -69,6 +70,8 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest
new LeaveGroupOperation
)

var heartbeatPurgatory: DelayedOperationPurgatory[DelayedHeartbeat] = _
var joinPurgatory: DelayedOperationPurgatory[DelayedJoin] = _
var groupCoordinator: GroupCoordinator = _

@Before
Expand All @@ -86,8 +89,8 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest

val config = KafkaConfig.fromProps(serverProps)

val heartbeatPurgatory = new DelayedOperationPurgatory[DelayedHeartbeat]("Heartbeat", timer, config.brokerId, reaperEnabled = false)
val joinPurgatory = new DelayedOperationPurgatory[DelayedJoin]("Rebalance", timer, config.brokerId, reaperEnabled = false)
heartbeatPurgatory = new DelayedOperationPurgatory[DelayedHeartbeat]("Heartbeat", timer, config.brokerId, reaperEnabled = false)
joinPurgatory = new DelayedOperationPurgatory[DelayedJoin]("Rebalance", timer, config.brokerId, reaperEnabled = false)

groupCoordinator = GroupCoordinator(config, zkClient, replicaManager, heartbeatPurgatory, joinPurgatory, timer.time, new Metrics())
groupCoordinator.startup(false)
Expand Down Expand Up @@ -124,6 +127,34 @@ class GroupCoordinatorConcurrencyTest extends AbstractCoordinatorConcurrencyTest
verifyConcurrentRandomSequences(createGroupMembers, allOperationsWithTxn)
}

@Test
def testConcurrentJoinGroupEnforceGroupMaxSize(): Unit = {
val groupMaxSize = 1
val newProperties = new Properties
newProperties.put(KafkaConfig.GroupMaxSizeProp, groupMaxSize.toString)
val config = KafkaConfig.fromProps(serverProps, newProperties)

if (groupCoordinator != null)
groupCoordinator.shutdown()
groupCoordinator = GroupCoordinator(config, zkClient, replicaManager, heartbeatPurgatory,
joinPurgatory, timer.time, new Metrics())
groupCoordinator.startup(false)

val members = new Group(s"group", nMembersPerGroup, groupCoordinator, replicaManager)
.members
val joinOp = new JoinGroupOperation()

verifyConcurrentActions(members.toSet.map(joinOp.actionNoVerify))

val errors = members.map { member =>
val joinGroupResult = joinOp.await(member, DefaultRebalanceTimeout)
joinGroupResult.error
}

assertEquals(groupMaxSize, errors.count(_ == Errors.NONE))
assertEquals(members.size-groupMaxSize, errors.count(_ == Errors.GROUP_MAX_SIZE_REACHED))
}

abstract class GroupOperation[R, C] extends Operation {
val responseFutures = new ConcurrentHashMap[GroupMember, Future[R]]()

Expand Down