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

merged 15 commits into from May 12, 2023

Conversation

jeffkbkim
Copy link
Contributor

@jeffkbkim jeffkbkim commented May 2, 2023 •
edited

Rewrites GroupMetadata as GenericGroup that will be used with the new group coordinator. Offset related fields, classes, and methods will not be included as they will be reworked as a separate offset manager component.

Committer Checklist (excluded from commit message)

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

@jeffkbkim jeffkbkim mentioned this pull request May 2, 2023
Merged
3 tasks
@dajac dajac added the KIP-848 label May 3, 2023
@jeffkbkim jeffkbkim marked this pull request as ready for review May 9, 2023 15:21
dajac
dajac reviewed May 10, 2023
View reviewed changes
Copy link
Contributor

@dajac dajac left a comment

Choose a reason for hiding this comment

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

@jeffkbkim Thanks for the patch. I made a first pass on it and left some small comments.

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Show resolved Hide resolved
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
...-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroupState.java Outdated Show resolved Hide resolved
...-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroupState.java Outdated Show resolved Hide resolved
...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java Outdated

@BeforeEach
public void initialize() {
group = new GenericGroup(new LogContext(), "groupId", Empty, Time.SYSTEM);
Copy link
Contributor

Choose a reason for hiding this comment

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

nit: I usually prefer to avoid global variable like this in tests but I leave it up to you.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

i'll keep this as is since this variable is used for all tests.

...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java Outdated Show resolved Hide resolved
rreddy-22
rreddy-22 reviewed May 10, 2023
View reviewed changes
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java
private GenericGroupState state;

/**
* The timestamp of when the group transitioned
Copy link
Contributor

Choose a reason for hiding this comment

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

nit:- of -> from*

Copy link
Contributor Author

Choose a reason for hiding this comment

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

this looks grammatically correct

rreddy-22
rreddy-22 reviewed May 10, 2023
View reviewed changes
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java
private final Map<String, String> staticMembers = new HashMap<>();

/**
* Members who have yet to (re)join the group
Copy link
Contributor

Choose a reason for hiding this comment

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

nit: have -> are*

Copy link
Contributor Author

Choose a reason for hiding this comment

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

this looks grammatically correct

Copy link
Contributor

Choose a reason for hiding this comment

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

oya yep mb

@jeffkbkim jeffkbkim mentioned this pull request May 11, 2023
Merged
3 tasks
dajac
dajac reviewed May 11, 2023
View reviewed changes
Copy link
Contributor

@dajac dajac left a comment

Choose a reason for hiding this comment

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

@jeffkbkim Thanks for the update. I left some comments for consideration.

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated
Comment on lines 310 to 314
public boolean isGenericGroup() {
return protocolType.map(type ->
type.equals(ConsumerProtocol.PROTOCOL_TYPE)
).orElse(false);
}
Copy link
Contributor

Choose a reason for hiding this comment

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

Was this method in the old implementation? The name is a bit weird because a generic group could use different protocol types (e.g. connect). Should it be named useConsumerGroupProtocol?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

got it. this is from
def isConsumerGroup: Boolean = protocolType.contains(ConsumerProtocol.PROTOCOL_TYPE)

the naming is confusing because we both consumer and generic groups but a generic group can expect a group using the consumer protocol

group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java Outdated Show resolved Hide resolved
...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java Outdated Show resolved Hide resolved
...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java Outdated Show resolved Hide resolved
...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java Show resolved Hide resolved
...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java Show resolved Hide resolved
...p-coordinator/src/test/java/org/apache/kafka/coordinator/group/generic/GenericGroupTest.java Outdated Show resolved Hide resolved
clolov
clolov reviewed May 11, 2023
View reviewed changes
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java
* The timestamp of when the group transitioned
* to its current state.
*/
private Optional<Long> currentStateTimestamp;
Copy link
Collaborator

@clolov clolov May 11, 2023 •
edited

Choose a reason for hiding this comment

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

Is this necessary to be an Optional? As far as I see we immediately define it in the constructor and we never set it equal to something which is empty. Am I missing something obvious?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

This is a bit awkward as the existing GroupMetadata updates this field when we read the group metadata record (https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/coordinator/group/GroupMetadataManager.scala#L1225).

So we should expect the new group metadata manager introduced in #13639 to perform this.

Copy link
Collaborator

Choose a reason for hiding this comment

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

Sorry, I do not understand. This field is currently private so the only way we can set it now is either in the constructor or via a setter. Do you mean that in the near future either we will make this field public or we will add a setter which will then be called by the GroupMetadataManager to possibly set this to an empty Optional? If so, then okay, this makes sense.

Copy link
Contributor

Choose a reason for hiding this comment

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

@clolov Your understanding is correct. The scala code doing this is here. We will need to add a setter at some point.

dajac
dajac approved these changes May 12, 2023
View reviewed changes
Copy link
Contributor

@dajac dajac left a comment

Choose a reason for hiding this comment

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

LGTM, thanks.

@dajac dajac merged commit cc011f7 into apache:trunk May 12, 2023
1 check failed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@dajac dajac dajac approved these changes

@clolov clolov clolov left review comments

@rreddy-22 rreddy-22 rreddy-22 left review comments

Assignees
No one assigned
Labels
KIP-848
Projects
None yet
Milestone
No milestone
4 participants
@jeffkbkim @dajac @clolov @rreddy-22