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
8 changes: 3 additions & 5 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 @@ -160,11 +160,9 @@ class GroupMetadataManager(brokerId: Int,

// 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) match {
case group => group != null && group.inLock(group.is(Dead))
}
})
isGroupLocal(groupId) && getGroup(groupId).forall { group =>
group.inLock(group.is(Dead))
}
}

// visible for testing
Expand Down
17 changes: 17 additions & 0 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 @@ -497,6 +497,23 @@ class GroupMetadataManagerTest {
}
}

@Test
def testGroupNotExists() {
// group is not owned
assertFalse(groupMetadataManager.groupNotExists(groupId))

groupMetadataManager.addPartitionOwnership(groupPartitionId)

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

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

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

Yes, I'll add that. Thanks!

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

// group is owned but not Dead
assertFalse(groupMetadataManager.groupNotExists(groupId))

group.transitionTo(Dead)
// group is owned and Dead
assertTrue(groupMetadataManager.groupNotExists(groupId))
}

private def appendConsumerOffsetCommit(buffer: ByteBuffer, baseOffset: Long, offsets: Map[TopicPartition, Long]) = {
val builder = MemoryRecords.builder(buffer, CompressionType.NONE, TimestampType.LOG_APPEND_TIME, baseOffset)
val commitRecords = createCommittedOffsetRecords(offsets)
Expand Down