KIP-229: DeleteGroups API by vahidhashemian · Pull Request #4479 · apache/kafka

vahidhashemian · 2018-01-26T19:36:36Z

This PR implements KIP-229.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

vahidhashemian · 2018-01-26T19:47:48Z

@hachikuji, due to the fast-approaching feature freeze, I thought I'd ask your opinion: The KIP proposes an error at the high level and then per group errors. But I now think that the high level error is not really something that could apply to all groups that are requested to be deleted. For example, Errors.COORDINATOR_NOT_AVAILABLE would apply to some (coordinator of some may be available while coordinator for some other may not), Errors.GROUP_AUTHORIZATION_FAILED would apply to some other, ... So, to me it sounds the high level code should be removed from the protocol.

Just wanted to get your feedback on my understanding is correct, and if so, the course of action. Thanks a lot.

hachikuji · 2018-01-26T19:54:50Z

@vahidhashemian Yes, I was in fact wondering about that when I read through the KIP. The only case I could think of when we'd take advantage of it would be unhandled errors. If we don't have a good use case for it at the moment, I think it would be fine to drop it.

vahidhashemian · 2018-01-26T19:58:02Z

@hachikuji thanks a lot for the quick response. Should I just update the KIP with this? Any notification/revote required?

hachikuji · 2018-01-26T19:58:53Z

I'd suggest updating the KIP and sending a message to the discussion thread. We often change minor details during implementation, so I don't think a revote will be needed, but we can see if anyone has feedback.

vahidhashemian · 2018-01-27T07:27:39Z

@hachikuji, would appreciate your feedback on this PR when you get a chance. Thanks!

hachikuji

Thanks for the patch. Left some comments.

hachikuji · 2018-01-29T21:32:14Z

+    @Override
+    public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        Errors error = Errors.forException(e);
+        Map<String, Errors> groupErrors = new HashMap<>();


nit: may as well initialize with the right size

hachikuji · 2018-01-29T21:35:48Z

+                return new DeleteGroupsResponse(throttleTimeMs, groupErrors);
+            default:
+                throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d",
+                    version(), this.getClass().getSimpleName(), ApiKeys.DELETE_GROUPS.latestVersion()));


nit: instead of the class name, maybe use ApiKeys.DELETE_GROUPS.name.

hachikuji · 2018-01-29T21:36:29Z

+
+
+    /**
+     * Possible error codes:


I think NOT_COORDINATOR should also be possible?

Correct. Also COORDINATOR_LOAD_IN_PROGRESS if I'm not mistaken?

hachikuji · 2018-01-29T23:06:40Z


+  def deleteConsumerGroups(groups: List[String]): Map[String, Errors] = {
+    var errors: Map[String, Errors] = Map()
+    val groupsPerCoordinator = groups.map { group =>


I think I'd suggest moving coordinator lookup to a separate function. You might also consider using the Either class to distinguish errors since the mixture of functional logic and updates to the mutable errors collection is a little odd.

I tried to improve upon this in the new commit. Please let me know what you think.

hachikuji · 2018-01-29T23:10:12Z

    }

+    override def deleteGroups(): Map[String, Errors] = {
+      val groupsToDelete = opts.options.valuesOf(opts.groupOpt).asScala.toList


Hmm.. It's a little weird that we allow multiple groups to be passed when using the new consumer, but we expect a single group for the old consumer. If we're to stay consistent, do you think it would be restrictive in practice to only support deletion of a single group at a time?

In the meantime I'll look at your other feedback (thanks btw) regarding this one, it seems the old consumer also supports deleting multiple groups, i.e. ... --delete --group group1 --group group2 works and attempts to remove both groups.

I originally wanted to support single group deletion only, but after considering the existing behavior for old consumer decided otherwise.

Ah, you are right. The name deleteForGroup is kind of misleading. Maybe it just needs to be pluralized. I wouldn't hate it if we came up with better names for all of these deleteForXXX APIs.

Sure, I gave this a quick try. Let me know if you have better suggestions.

hachikuji · 2018-01-29T23:16:13Z

+    var result: Map[String, Errors] = Map()
+
+    groupIds.foreach { groupId =>
+      if (!groupMetadataCache.contains(groupId))


This "check and act" is not safe since we're not holding a lock. It would be better to get the GroupMetadata object and check if it is null. If it is not null, then we need to grab the group lock before checking its state and attempting to delete its state.

That's correct. It seems because of this lock we need to delete groups one by one then (as in the new commit)?

hachikuji · 2018-01-29T23:17:52Z

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


I don't think passing None works. Looking at cleanupGroupMetadata, that would just result in removal of the expired offsets.

Correct, but since we pass Long.MaxValue as the current time, all offsets in the passed groups expire. Would that work?

hachikuji · 2018-01-29T23:19:03Z

-  def cleanupGroupMetadata(deletedTopicPartitions: Option[Seq[TopicPartition]]) {
-    val startMs = time.milliseconds()
+  def cleanupGroupMetadata(deletedTopicPartitions: Option[Seq[TopicPartition]],
+                           groups: Iterable[GroupMetadata] = groupMetadataCache.values,


It would be better not to have optional arguments. Let's make the caller provide the values explicitly.

hachikuji · 2018-01-29T23:21:33Z

+    groups.foreach { group =>
+      if (!authorize(request.session, Delete, new Resource(Group, group))) {
+        unauthorizedGroupsDeletionResult += (group -> Errors.GROUP_AUTHORIZATION_FAILED)
+        groups -= group


Perhaps we can partition to split the incoming group list into the authorized and unauthorized groups.

Thanks for the suggestion, makes a lot of sense.

hachikuji · 2018-01-29T23:23:27Z

+  }
+
+  @Test
+  def testDeleteEmptyGroup() {


We should have a test case which tests removal when there are stored offsets.

I added one in the new commit.

vahidhashemian

@hachikuji thanks for the feedback. I tried to address them in the new commit.

vahidhashemian · 2018-01-30T00:23:43Z

+
+
+    /**
+     * Possible error codes:


Correct. Also COORDINATOR_LOAD_IN_PROGRESS if I'm not mistaken?

vahidhashemian · 2018-01-30T01:00:07Z


+  def deleteConsumerGroups(groups: List[String]): Map[String, Errors] = {
+    var errors: Map[String, Errors] = Map()
+    val groupsPerCoordinator = groups.map { group =>


I tried to improve upon this in the new commit. Please let me know what you think.

vahidhashemian · 2018-01-30T01:05:02Z

      else if (opts.options.has(opts.topicOpt))
        deleteAllForTopic()
+
+      Map()


Sounds good. I updated this in the new commit.

vahidhashemian · 2018-01-30T01:20:20Z

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


Correct, but since we pass Long.MaxValue as the current time, all offsets in the passed groups expire. Would that work?

vahidhashemian · 2018-01-30T01:21:37Z

+    var result: Map[String, Errors] = Map()
+
+    groupIds.foreach { groupId =>
+      if (!groupMetadataCache.contains(groupId))


That's correct. It seems because of this lock we need to delete groups one by one then (as in the new commit)?

vahidhashemian · 2018-01-30T01:35:06Z

+    groups.foreach { group =>
+      if (!authorize(request.session, Delete, new Resource(Group, group))) {
+        unauthorizedGroupsDeletionResult += (group -> Errors.GROUP_AUTHORIZATION_FAILED)
+        groups -= group


Thanks for the suggestion, makes a lot of sense.

vahidhashemian · 2018-01-30T06:16:53Z

    }

+    override def deleteGroups(): Map[String, Errors] = {
+      val groupsToDelete = opts.options.valuesOf(opts.groupOpt).asScala.toList


Sure, I gave this a quick try. Let me know if you have better suggestions.

vahidhashemian · 2018-01-30T06:31:23Z

+  }
+
+  @Test
+  def testDeleteEmptyGroup() {


I added one in the new commit.

omkreddy

LGTM

omkreddy · 2018-01-30T09:24:17Z

+      authorize(request.session, Delete, new Resource(Group, group))
+    }
+
+    val groupDeletionResult = groupCoordinator.handleDeleteGroups(authorizedGroups)._2 ++


looks like we are ignoring groupCoordinator.handleDeleteGroups(authorizedGroups)._1 error here.
handleDeleteGroups(authorizedGroups) can return Errors.COORDINATOR_NOT_AVAILABLE.

Thanks for catching this. I missed updating this with the recent change to the protocol. Will update it in the next commit.

hachikuji

Left a few more comments. I think we still need a little work to make the deletion safe for edge cases around coordinator failover.

hachikuji · 2018-01-30T17:02:37Z

-    }.filter(_._1 != null)
+    }
+
+    val groupCoordinator = groups.map(group => (group -> coordinatorLookup(group)))


Seems this is unused

hachikuji · 2018-01-30T17:09:05Z

+          errors += (group -> error)
+        case Left(coordinator) =>
+          groupsPerCoordinator.get(coordinator) match {
+            case Some(gList: List[String]) =>


nit: I don't think you need the type.

hachikuji · 2018-01-30T17:11:55Z

+      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))


nit: unneeded parenthesis around response.hasError(group)

hachikuji · 2018-01-30T17:17:16Z

-        deleteAllForTopic()
+        deleteAllGroupsInfoForTopic()
+
+      Map()


Seems like you were intending to use the results of the deleteGroupsInfo and such. We should probably have a test case (could be done in a follow-up).

Sure, I'll submit a separate PR with proper test(s) after this is merged.

hachikuji · 2018-01-30T17:18:53Z

+          if (AdminUtils.deleteConsumerGroupInfoForTopicInZK(zkUtils, group, topic)) {
            println(s"Deleted consumer group information for group '$group' topic '$topic' in zookeeper.")
-          else
+            (group -> Errors.NONE)


nit: unneeded parenthesis. A few more like this.

hachikuji · 2018-01-30T17:31:29Z

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


hachikuji · 2018-01-30T18:44:03Z

+      groupIds.foreach { groupId =>
+        if (!validGroupId(groupId))
+          groupErrors += (groupId -> Errors.INVALID_GROUP_ID)
+        else if (!isCoordinatorForGroup(groupId)) {


nit: looks weird that only this branch has braces

hachikuji · 2018-01-30T19:23:30Z

+      var groupErrors: Map[String, Errors] = Map()
+      var eligibleGroups: Seq[String] = Seq()
+
+      groupIds.foreach { groupId =>


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:

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?

Check if the group is not empty. If so, return the GROUP_NOT_EMPTY error code.

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.

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

It seems all this can be done here and we could get rid of GroupMetadataManager.deleteGroups(). Do you see an issue with it?

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.

Seems reasonable to me.

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.

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

vahidhashemian

@hachikuji thanks for another review. Could you please clarify on a couple of questions inline before I submit another patch? Thanks.

vahidhashemian · 2018-01-30T19:39:34Z

-        deleteAllForTopic()
+        deleteAllGroupsInfoForTopic()
+
+      Map()


Sure, I'll submit a separate PR with proper test(s) after this is merged.

vahidhashemian · 2018-01-30T20:47:11Z

+      var groupErrors: Map[String, Errors] = Map()
+      var eligibleGroups: Seq[String] = Seq()
+
+      groupIds.foreach { groupId =>


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

It seems all this can be done here and we could get rid of GroupMetadataManager.deleteGroups(). Do you see an issue with it?

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

Thank for the updates. A few more comments.

hachikuji · 2018-01-30T22:13:15Z

+        else {
+          groupManager.getGroup(groupId) match {
+            case None =>
+              groupErrors += groupId -> Errors.GROUP_ID_NOT_FOUND


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

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.

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.

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).

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.

Aah, right. That makes sense.

hachikuji · 2018-01-30T22:14:43Z

-  }
-
-  @Test
-  def testDeleteNonEmptyGroup() {


Why remove these test cases?

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

hachikuji · 2018-01-30T22:20:49Z

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


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?

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 :)

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.

hachikuji · 2018-01-31T16:22:08Z

@vahidhashemian If you can update the patch this morning, we may still be able to get it into this release. The main thing from my perspective is ensuring that the GROUP_ID_NOT_FOUND error code is returned correctly as discussed above.

vahidhashemian · 2018-01-31T16:40:10Z

@hachikuji I just updated the patch, without the improvement on cleanupGroupMetadata. I can work on it in the meantime, and submit a patch separately (perhaps under a different PR).

hachikuji · 2018-01-31T16:43:08Z


+  // 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))


Should have mentioned before, but we do need to grab the group lock to check the state.

Correct, thanks for catching. Hopefully the new commit works.

hachikuji · 2018-01-31T18:23:45Z

  // 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))
+    isGroupLocal(groupId) && (!groupMetadataCache.contains(groupId) || {


Can you write a short test case to make sure this function works correctly. Also, I think this is a bit more concise:

isGroupLocal(groupId) && getGroup(groupId).forall { group => group.inLock(group.is(Dead)) }

Thanks for the code improvement suggestion. I added a basic unit test in the new commit.

hachikuji · 2018-01-31T19:00:43Z

+    // group is not owned
+    assertFalse(groupMetadataManager.groupNotExists(groupId))
+
+    groupMetadataManager.addPartitionOwnership(groupPartitionId)


Following this and prior to adding the group, we should see groupNotExists return true?

Yes, I'll add that. Thanks!

hachikuji

LGTM. Thanks for the patch!

hachikuji · 2018-01-31T21:19:19Z

The test failures appear unrelated. Merging to trunk.

vahidhashemian · 2018-01-31T21:26:37Z

Great, and thanks for quick reviews!

asfgit · 2018-01-31T23:26:20Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-test-coverage/203/

vahidhashemian force-pushed the KAFKA-6275 branch 3 times, most recently from 8dd6874 to ae247cf Compare January 27, 2018 00:58

KIP-229: DeleteGroups API (WIP)

b282a21

vahidhashemian force-pushed the KAFKA-6275 branch from ae247cf to b282a21 Compare January 27, 2018 05:40

vahidhashemian changed the title ~~KIP-229: DeleteGroups API (WIP)~~ KIP-229: DeleteGroups API Jan 27, 2018

hachikuji self-assigned this Jan 28, 2018

jeqo mentioned this pull request Jan 28, 2018

KIP-222 - Add Consumer Group operations to Admin API #4454

Closed

hachikuji reviewed Jan 29, 2018

View reviewed changes

Addressed the first round of feedback

10aab1a

vahidhashemian commented Jan 30, 2018

View reviewed changes

Remove the top level error code in GroupCoordinator

6d3a0b7

omkreddy reviewed Jan 30, 2018

View reviewed changes

Consolidated the broker side deletion logic into GroupCoordinator

bcad73a

vahidhashemian force-pushed the KAFKA-6275 branch from d37d459 to 6d3a0b7 Compare January 30, 2018 17:23

hachikuji reviewed Jan 30, 2018

View reviewed changes

vahidhashemian commented Jan 30, 2018

View reviewed changes

hachikuji reviewed Jan 30, 2018

View reviewed changes

Improved the checks for GROUP_ID_NOT_FOUND

a602d3c

hachikuji reviewed Jan 31, 2018

View reviewed changes

Acquired the group lock before checking group state

8fb935f

hachikuji reviewed Jan 31, 2018

View reviewed changes

Added a unit test for groupNotExists

eabc8b3

hachikuji reviewed Jan 31, 2018

View reviewed changes

Added one more assertion in the new unit test

ca60b65

hachikuji approved these changes Jan 31, 2018

View reviewed changes

hachikuji merged commit 1ed6da7 into apache:trunk Jan 31, 2018

Conversation

vahidhashemian commented Jan 26, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

vahidhashemian commented Jan 26, 2018

Uh oh!

hachikuji commented Jan 26, 2018

Uh oh!

vahidhashemian commented Jan 26, 2018

Uh oh!

hachikuji commented Jan 26, 2018

Uh oh!

vahidhashemian commented Jan 27, 2018

Uh oh!

hachikuji left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vahidhashemian left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

omkreddy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vahidhashemian commented Jan 26, 2018 •

edited

Loading