Skip to content

KIP-229: DeleteGroups API #4479

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 8 commits into from
Jan 31, 2018
Merged

KIP-229: DeleteGroups API #4479

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
b282a21
KIP-229: DeleteGroups API (WIP)
vahidhashemian Jan 26, 2018
10aab1a
Addressed the first round of feedback
vahidhashemian Jan 30, 2018
6d3a0b7
Remove the top level error code in GroupCoordinator
vahidhashemian Jan 30, 2018
bcad73a
Consolidated the broker side deletion logic into GroupCoordinator
vahidhashemian Jan 30, 2018
a602d3c
Improved the checks for GROUP_ID_NOT_FOUND
vahidhashemian Jan 31, 2018
8fb935f
Acquired the group lock before checking group state
vahidhashemian Jan 31, 2018
eabc8b3
Added a unit test for groupNotExists
vahidhashemian Jan 31, 2018
ca60b65
Added one more assertion in the new unit test
vahidhashemian Jan 31, 2018
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
14 changes: 6 additions & 8 deletions core/src/main/scala/kafka/admin/AdminClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -383,22 +383,20 @@ class AdminClient(val time: Time,
}
}

val groupCoordinator = groups.map(group => (group -> coordinatorLookup(group)))

var errors: Map[String, Errors] = Map()
var groupsPerCoordinator: Map[Node, List[String]] = Map()

groups.foreach { group =>
coordinatorLookup(group) match {
case Right(error) =>
errors += (group -> error)
errors += group -> error
case Left(coordinator) =>
groupsPerCoordinator.get(coordinator) match {
case Some(gList: List[String]) =>
case Some(gList) =>
val gListNew = group :: gList
groupsPerCoordinator += (coordinator -> gListNew)
groupsPerCoordinator += coordinator -> gListNew
case None =>
groupsPerCoordinator += (coordinator -> List(group))
groupsPerCoordinator += coordinator -> List(group)
}
}
}
Expand All @@ -407,8 +405,8 @@ class AdminClient(val time: Time,
val responseBody = send(coordinator, ApiKeys.DELETE_GROUPS, new DeleteGroupsRequest.Builder(groups.toSet.asJava))
val response = responseBody.asInstanceOf[DeleteGroupsResponse]
groups.foreach {
case group if (response.hasError(group)) => errors += (group -> response.errors.get(group))
case group => errors += (group -> Errors.NONE)
case group if response.hasError(group) => errors += group -> response.errors.get(group)
case group => errors += group -> Errors.NONE
}
}

Expand Down
14 changes: 7 additions & 7 deletions core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -480,17 +480,17 @@ object ConsumerGroupCommand extends Logging {
try {
if (AdminUtils.deleteConsumerGroupInZK(zkUtils, group)) {
println(s"Deleted all consumer group information for group '$group' in zookeeper.")
(group -> Errors.NONE)
group -> Errors.NONE
}
else {
printError(s"Delete for group '$group' failed because its consumers are still active.")
(group -> Errors.NON_EMPTY_GROUP)
group -> Errors.NON_EMPTY_GROUP
}
}
catch {
case e: ZkNoNodeException =>
printError(s"Delete for group '$group' failed because group does not exist.", Some(e))
(group -> Errors.forException(e))
group -> Errors.forException(e)
}
}.toMap
}
Expand All @@ -503,17 +503,17 @@ object ConsumerGroupCommand extends Logging {
try {
if (AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkUtils, group, topic)) {
println(s"Deleted consumer group information for group '$group' topic '$topic' in zookeeper.")
(group -> Errors.NONE)
group -> Errors.NONE
}
else {
printError(s"Delete for group '$group' topic '$topic' failed because its consumers are still active.")
(group -> Errors.NON_EMPTY_GROUP)
group -> Errors.NON_EMPTY_GROUP
}
}
catch {
case e: ZkNoNodeException =>
printError(s"Delete for group '$group' topic '$topic' failed because group does not exist.", Some(e))
(group -> Errors.forException(e))
group -> Errors.forException(e)
}
}.toMap
}
Expand All @@ -523,7 +523,7 @@ object ConsumerGroupCommand extends Logging {
Topic.validate(topic)
val deletedGroups = AdminUtils.deleteAllConsumerGroupInfoForTopicInZK(zkUtils, topic)
println(s"Deleted consumer group information for all inactive consumer groups for topic '$topic' in zookeeper.")
deletedGroups.map((_, Errors.NONE)).toMap
deletedGroups.map(_ -> Errors.NONE).toMap

}

Expand Down
45 changes: 35 additions & 10 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 @@ -340,23 +340,48 @@ class GroupCoordinator(val brokerId: Int,

def handleDeleteGroups(groupIds: Set[String]): Map[String, Errors] = {
if (!isActive.get) {
groupIds.map((_ -> Errors.COORDINATOR_NOT_AVAILABLE)).toMap
groupIds.map(_ -> Errors.COORDINATOR_NOT_AVAILABLE).toMap
} else {
var groupErrors: Map[String, Errors] = Map()
var eligibleGroups: Seq[String] = Seq()
var eligibleGroups: Seq[GroupMetadata] = Seq()

groupIds.foreach { groupId =>

@hachikuji hachikuji Jan 30, 2018

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.

In fact, this is also a "check and act." It is possible for an eligible group to be unloaded between these checks and the call to deleteGroups.

I think we should follow a structure more similar to the other API handlers. I would suggest moving the state checking that we currently have in GroupMetadataManager.deleteGroups into the else case below. We should do the following:

  1. Check if there is no group or if the group is Dead. If so, it could mean that it has already been moved to another broker or it could mean that the group doesn't exist. I am not sure we have a bulletproof way to distinguish these cases, but maybe we could just check again if the coordinator is still correct?
  2. Check if the group is not empty. If so, return the GROUP_NOT_EMPTY error code.
  3. If the group is empty, we should transition to Dead. Once we do so, we are ensured that no other thread will attempt to use the GroupMetadata object. We can then collect this eligible group in a collection (as is currently done) and send it to cleanupGroupMetadata outside of the lock.

@vahidhashemian vahidhashemian Jan 30, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I'm working on this and have a couple of questions for now:

  1. It seems all this can be done here and we could get rid of GroupMetadataManager.deleteGroups(). Do you see an issue with it?
  2. Could you please clarify what you mean by "check again if the coordinator is still correct" when group cannot be found or is Dead?

Thanks.

@hachikuji hachikuji Jan 30, 2018 •
edited
Loading

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.

  1. Seems reasonable to me.

  2. We are trying to address the case in which a group gets deleted or migrated in between the time that we check if the coordinator is assigned and the time we delete the group metadata. We have the Dead state for this purpose, so whenever we check GroupMetadata, the first thing we should check is whether it is already Dead. If it is, then we know it was either already deleted or already migrated. My suggestion is to check again whether we are still the coordinator for the group to disambiguate the two cases.

Note that I am not sure that this is 100% bulletproof. For example, it may not handle the case when the coordinator is migrated away and then back very quickly. A spurious NOT_COORDINATOR error is not a big deal because clients are expected to handle it, but I am not too sure about the GROUP_ID_NOT_FOUND error. Maybe clients just have to treat it with the same skepticism that they treat the UNKOWN_TOPIC_OR_PARTITION errors.

@vahidhashemian vahidhashemian Jan 30, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Thanks for clarifying this. I'll try to do just that in the new patch. Will submit shortly.

if (!validGroupId(groupId))
groupErrors += (groupId -> Errors.INVALID_GROUP_ID)
else if (!isCoordinatorForGroup(groupId)) {
groupErrors += (groupId -> Errors.NOT_COORDINATOR)
} else if (isCoordinatorLoadInProgress(groupId))
groupErrors += (groupId -> Errors.COORDINATOR_LOAD_IN_PROGRESS)
else
eligibleGroups ++= Seq(groupId)
groupErrors += groupId -> Errors.INVALID_GROUP_ID
else if (!isCoordinatorForGroup(groupId))
groupErrors += groupId -> Errors.NOT_COORDINATOR
else if (isCoordinatorLoadInProgress(groupId))
groupErrors += groupId -> Errors.COORDINATOR_LOAD_IN_PROGRESS
else {
groupManager.getGroup(groupId) match {
case None =>
groupErrors += groupId -> Errors.GROUP_ID_NOT_FOUND

@hachikuji hachikuji Jan 30, 2018

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.

This case should be handled the same as if the group is dead. You can probably add a little helper to avoid the duplication.

@hachikuji hachikuji Jan 30, 2018 •
edited
Loading

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.

On second thought, this probably doesn't solve the underlying issue. I'm trying to think how we can be sure that we're returning this error code correctly. Maybe we need to check for group existence while holding the ownedPartitions lock in GroupMetadataManager.

@hachikuji hachikuji Jan 30, 2018

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.

If ownedPartitions includes the corresponding topic partition for the group, and if the cached group either doesn't exist or is Dead, then I think it is safe to return GROUP_ID_NOT_FOUND. Maybe we can just add a method like the following to GroupMetadataManager:

// return true iff group is owned and the group doesn't exist
def groupNotExists(groupId: String) = inLock(partitionLock) {
  isGroupLocal(groupId) && (!groupMetadataCache.contains(groupId) || groupMetadataCache.get(groupId).is(Dead))
}

Then we can use this function here instead of checking the coordinator again. The name could probably be improved.

@vahidhashemian vahidhashemian Jan 30, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

For both case None and case Dead we already know the second part ((!groupMetadataCache.contains(groupId) || groupMetadataCache.get(groupId).is(Dead))) is true. So, it suffices to check isGroupLocal(groupId) (to avoid redundant checks). Is that correct? If so, we wouldn't need this helper (at least here).

@hachikuji hachikuji Jan 30, 2018

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.

The point is to check it while holding the partition lock so that it is an atomic operation. This ensures that we will not have any race conditions with partition loading/unloading.

@vahidhashemian vahidhashemian Jan 30, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Aah, right. That makes sense.

case Some(group) =>
group.inLock {
group.currentState match {
case Dead =>
if (!isCoordinatorForGroup(groupId))
groupErrors += groupId -> Errors.NOT_COORDINATOR
else
groupErrors += groupId -> Errors.GROUP_ID_NOT_FOUND
case Empty =>
group.transitionTo(Dead)
eligibleGroups :+= group
case _ =>
groupErrors += groupId -> Errors.NON_EMPTY_GROUP
}
}
}
}
}

if (eligibleGroups.nonEmpty) {
groupManager.cleanupGroupMetadata(None, eligibleGroups, Long.MaxValue)

@hachikuji hachikuji Jan 30, 2018

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.

This still feels a bit hacky. As an alternative, maybe we can let the offset selector be provided as a function. Something like this:

def cleanupGroupMetadata(
 groups: Iterable[GroupMetadata], 
 collectOffsetsToRemove: Group => Map[TopicPartition, OffsetAndMetadata])

What do you think?

@vahidhashemian vahidhashemian Jan 31, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I'm not sure which part you consider hacky, and am trying to understand your suggestion.

For the sake of deleteGroups functionality, we can use group.allOffsets that conforms to the function signature above. But how about the existing functionality, where we want to delete specific topic partitions from a group: groupManager.cleanupGroupMetadata(Some(topicPartitions), groupManager.currentGroups, time.milliseconds()) and populate the corresponding OffsetAndMetadata values? I'm assuming we want to reuse the same cleanupGroupMetadata method for both cases.

On the same assumption, we also need to factor in the concept of current time so we can determine the expired offsets for the existing functionality.

On the other hand if you are proposing to create A new cleanupGroupMetadata method that calls on the existing method, we should make this call once per group (since topic partitions are group-specific).

Or maybe I'm missing the point :)

@hachikuji hachikuji Jan 31, 2018

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.

It's not that big of a deal. I just thought it was a mild abuse to reuse the expiration logic to delete all offsets. Alternatively, what I was suggesting is to let the caller choose the offsets to delete.

groupErrors ++= eligibleGroups.map(_.groupId -> Errors.NONE).toMap
info(s"The following groups were deleted: ${eligibleGroups.map(_.groupId).mkString(", ")}")
}

groupErrors ++ groupManager.deleteGroups(eligibleGroups)
groupErrors
}
}

Expand Down
31 changes: 0 additions & 31 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 @@ -184,37 +184,6 @@ class GroupMetadataManager(brokerId: Int,
}
}

/**
* Delete groups
*/
def deleteGroups(groupIds: Seq[String]): Map[String, Errors] = {
var eligibleGroups: Seq[GroupMetadata] = Seq()
var result: Map[String, Errors] = Map()

groupIds.foreach { groupId =>
getGroup(groupId) match {
case None =>
result += (groupId -> Errors.GROUP_ID_NOT_FOUND)
case Some(group) =>
group.inLock {
if (!group.is(Empty))
result += (groupId -> Errors.NON_EMPTY_GROUP)
else {
eligibleGroups :+= group
cleanupGroupMetadata(None, Seq(group), Long.MaxValue)
}
}
}
}

if (eligibleGroups.nonEmpty) {
result ++= eligibleGroups.map(_.groupId -> Errors.NONE).toMap
info(s"The following groups were deleted: ${eligibleGroups.map(_.groupId).mkString(", ")}")
}

result
}

def storeGroup(group: GroupMetadata,
groupAssignment: Map[String, Array[Byte]],
responseCallback: Errors => Unit): Unit = {
Expand Down
2 changes: 1 addition & 1 deletion core/src/main/scala/kafka/server/KafkaApis.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1230,7 +1230,7 @@ class KafkaApis(val requestChannel: RequestChannel,
}

val groupDeletionResult = groupCoordinator.handleDeleteGroups(authorizedGroups) ++
unauthorizedGroups.map((_ -> Errors.GROUP_AUTHORIZATION_FAILED))
unauthorizedGroups.map(_ -> Errors.GROUP_AUTHORIZATION_FAILED)

sendResponseMaybeThrottle(request, requestThrottleMs =>
new DeleteGroupsResponse(requestThrottleMs, groupDeletionResult.asJava))
Expand Down
22 changes: 0 additions & 22 deletions core/src/test/scala/unit/kafka/coordinator/group/GroupMetadataManagerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -714,28 +714,6 @@ class GroupMetadataManagerTest {
assertEquals(group, groupMetadataManager.addGroup(new GroupMetadata("foo", initialState = Empty)))
}

@Test
def testDeleteEmptyGroup() {
val group = new GroupMetadata(groupId, initialState = Empty)
assertEquals(group, groupMetadataManager.addGroup(group))

EasyMock.expect(replicaManager.getMagic(EasyMock.anyObject())).andStubReturn(Some(RecordBatch.CURRENT_MAGIC_VALUE))
mockGetPartition()
EasyMock.replay(replicaManager, partition)

val result = groupMetadataManager.deleteGroups(Seq(group.groupId))
assert(result.size == 1 && result.contains(group.groupId) && result.get(group.groupId).contains(Errors.NONE))
}

@Test
def testDeleteNonEmptyGroup() {

@hachikuji hachikuji Jan 30, 2018

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.

Why remove these test cases?

@vahidhashemian vahidhashemian Jan 30, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

They made a call to GroupMetadataManager.deleteGroups(...) that we just deleted. Similar tests exist in GroupCoordinatorTest.

val group = new GroupMetadata(groupId, initialState = Stable)
assertEquals(group, groupMetadataManager.addGroup(group))

val result = groupMetadataManager.deleteGroups(Seq(group.groupId))
assert(result.size == 1 && result.contains(group.groupId) && result.get(group.groupId).contains(Errors.NON_EMPTY_GROUP))
}

@Test
def testStoreEmptyGroup() {
val generation = 27
Expand Down