Skip to content
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

Jump to bottom

KAFKA-14500; [2/N] Rewrite GroupMetadata in Java #13663

Merged
merged 15 commits into from
May 12, 2023
Merged

KAFKA-14500; [2/N] Rewrite GroupMetadata in Java #13663

Changes from 1 commit
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
d7113b5
rewrite GroupMetadata
jeffkbkim May 2, 2023
fc0b7b4
fix
jeffkbkim May 2, 2023
6b15b2d
javadocs
jeffkbkim May 5, 2023
ca3db3e
add unit test
jeffkbkim May 6, 2023
aabc26e
remove offset classes
jeffkbkim May 8, 2023
a0351b3
don't handle exception in future completion
jeffkbkim May 8, 2023
8f1ce91
rebase on trunk
jeffkbkim May 9, 2023
dd24c78
undo
jeffkbkim May 9, 2023
785cc11
checkstyle
jeffkbkim May 9, 2023
e5bd48e
address comments
jeffkbkim May 10, 2023
461f785
add group id getter
jeffkbkim May 10, 2023
6cde9b3
add protocol type getter
jeffkbkim May 10, 2023
6f65670
address comments
jeffkbkim May 11, 2023
1238530
address comments
jeffkbkim May 11, 2023
c47b4a6
address comments
jeffkbkim May 11, 2023
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
Prev Previous commit
Next Next commit
fix
  • Loading branch information
@jeffkbkim
jeffkbkim committed May 9, 2023
commit fc0b7b4da60a9ba9137cd7eb983e59c9baf60d9a
5 changes: 2 additions & 3 deletions core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -451,7 +451,7 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
val (protocol, _) = allMemberMetadata
.map(_.vote(candidates))
.groupBy(identity)
.maxBy { case (a, votes) => votes.size }
.maxBy { case (_, votes) => votes.size }

protocol
}
Expand Down Expand Up @@ -507,8 +507,7 @@ private[group] class GroupMetadata(val groupId: String, initialState: GroupState
// future versions of the consumer protocol. VO must prefix all new versions.
val buffer = ByteBuffer.wrap(member.metadata(protocolName.get))
ConsumerProtocol.deserializeVersion(buffer)
val a = ConsumerProtocol.deserializeSubscription(buffer, 0).topics.asScala.toSet
a
ConsumerProtocol.deserializeSubscription(buffer, 0).topics.asScala.toSet
}.reduceLeft(_ ++ _)
)
} catch {
Expand Down