You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "apoorvmittal10 (via GitHub)" <gi...@apache.org> on 2024/05/23 19:12:42 UTC
[PR] KAFKA-16822: Abstract consumer group to share functionality with share group [kafka]
apoorvmittal10 opened a new pull request, #16054:
URL: https://github.com/apache/kafka/pull/16054
Abstracted code for 2 classes `ConsumerGroup` and `ConsumerGroupMember` to `AbstractGroup` and `GroupMember` respectively. The new abstract classes are created to share common functionality with `ShareGroup` and `ShareGroupMember` which are being introduced with KIP-932.
The PR is majorly code refactoring from existing classes to abstract classes. Also created a new package called `common` where `MemberState` class is moved, in upcoming PRs will move common classes for `Share` and `Consumer` Group in `common` package itself.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613579324
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
The protocol behind the groups seems to be the main distinction as I see it (meaning the base protocol). Classic groups as part of the classic protocol, vs share/consumer/streams.. that all build on top of the new protocol. Other than that, the groups are quite different I would say, so hard to find a common feature. Seeing it like this, the non-classic seems like a sensible umbrella term (given that coincidently, "classic" is the term we are standardizing for the old protocol and group, so convenient hint to the protocol too).
No matter the name, I would strongly suggest we add a the java doc for this class btw.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1626215818
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/AbstractModernGroup.java:
##########
@@ -0,0 +1,713 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.modern;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.Group;
+import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
+import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
+import org.apache.kafka.coordinator.group.Utils;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+/**
+ * The abstract group provides definitions for the consumer and share group.
+ */
+public abstract class AbstractModernGroup<T extends ModernGroupMember> implements Group {
Review Comment:
I think your suggestion is better and I am inclined to remove the `abstract` prefix. I have done that and changed the name to `ModerGroup` and kept `ModernGroupMember` as it is.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1617011178
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMember.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.MemberState;
+import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Abstract member common for group members.
+ */
+public abstract class GroupMember {
Review Comment:
Moved it to `modern` package as discussed above with renamed as `ModernGroupMember`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1626207415
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java:
##########
@@ -166,4 +166,20 @@ void validateOffsetFetch(
* @return true if the state includes, false otherwise.
*/
boolean isInStates(Set<String> statesFilter, long committedOffset);
+
+ /**
+ * Returns true if the member exists.
+ *
+ * @param memberId The member id.
+ *
+ * @return A boolean indicating whether the member exists or not.
+ */
+ boolean hasMember(String memberId);
+
+ /**
+ * Returns number of members in the group.
+ *
+ * @return The number of members.
+ */
+ int numMembers();
Review Comment:
Yeah may be, but just wanted to have minimal changes in refactoring and re-used existing i.e. ConsumerGroup already had `numMembers()` which I used for Classic Group as well. So I ll leave this like that for now. Incase we need to do that then will open minor refactor PR which has specific changes for renaming it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1626048896
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/AbstractModernGroup.java:
##########
@@ -0,0 +1,713 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.modern;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.Group;
+import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
+import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
+import org.apache.kafka.coordinator.group.Utils;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+/**
+ * The abstract group provides definitions for the consumer and share group.
+ */
+public abstract class AbstractModernGroup<T extends ModernGroupMember> implements Group {
Review Comment:
We use the Abstract prefix for the group class but not for the member class in the same situation. What about simply having a `ModernGroup` and `ModernGroupMember`? I find it more consistent and easy to reason about. (I personally prefer to loose the Abstract in the name, but if we prefer to keep it, I would then suggest we add it to the member class too)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1617010639
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
Review Comment:
Moved it to `modern` package as discussed above with renamed as `AbstractModernGroup`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613594739
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
My idea about AbstractGroup was that down the line AbstractGroup would be the one extended if we require additional groups and classic group is just an exception for now for backward compatibilty.
@lianetm I agree with java doc, will add.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1622198904
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java:
##########
@@ -166,4 +166,20 @@ void validateOffsetFetch(
* @return true if the state includes, false otherwise.
Review Comment:
Didn't move it as the interface is being used in ClassicGroup as well, hence just moved the one's which are common for Share and Comsumer group to `modern`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613638293
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
One way to address this would be to use the following structure:
* coordinator.classic.* for classic group related classes
* coordinator.modern.* (or nextgen) for common classes used by next gen groups
* coordinator.modern.consumer.* for specific classes
* coordinator.modern.share.*
* ...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1626016857
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java:
##########
@@ -166,4 +166,20 @@ void validateOffsetFetch(
* @return true if the state includes, false otherwise.
*/
boolean isInStates(Set<String> statesFilter, long committedOffset);
+
+ /**
+ * Returns true if the member exists.
+ *
+ * @param memberId The member id.
+ *
+ * @return A boolean indicating whether the member exists or not.
+ */
+ boolean hasMember(String memberId);
+
+ /**
+ * Returns number of members in the group.
+ *
+ * @return The number of members.
+ */
+ int numMembers();
Review Comment:
nit: `memberCount` clearer maybe? (I instantly think of partitionCount used elsewhere, sounds clearer to me)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1625772953
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/AbstractModernGroup.java:
##########
@@ -0,0 +1,713 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.modern;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.Group;
+import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
+import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
+import org.apache.kafka.coordinator.group.Utils;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+/**
+ * The abstract group provides definitions for the consumer and share group.
+ */
+public abstract class AbstractModernGroup<T extends ModernGroupMember> implements Group {
+
+ public static class DeadlineAndEpoch {
+ static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0);
+
+ public final long deadlineMs;
+ public final int epoch;
+
+ DeadlineAndEpoch(long deadlineMs, int epoch) {
+ this.deadlineMs = deadlineMs;
+ this.epoch = epoch;
+ }
+ }
+
+ /**
+ * The snapshot registry.
+ */
+ protected final SnapshotRegistry snapshotRegistry;
+
+ /**
+ * The group id.
+ */
+ protected final String groupId;
+
+ /**
+ * The group epoch. The epoch is incremented whenever the subscriptions
+ * are updated and it will trigger the computation of a new assignment
+ * for the group.
+ */
+ protected final TimelineInteger groupEpoch;
+
+ /**
+ * The group members.
+ */
+ protected final TimelineHashMap<String, T> members;
+
+ /**
+ * The number of subscribers per topic.
+ */
+ protected final TimelineHashMap<String, Integer> subscribedTopicNames;
+
+ /**
+ * The metadata associated with each subscribed topic name.
+ */
+ protected final TimelineHashMap<String, TopicMetadata> subscribedTopicMetadata;
+
+ /**
+ * The group's subscription type.
+ * This value is set to Homogeneous by default.
+ */
+ protected final TimelineObject<SubscriptionType> subscriptionType;
+
+ /**
+ * The target assignment epoch. An assignment epoch smaller than the group epoch
+ * means that a new assignment is required. The assignment epoch is updated when
+ * a new assignment is installed.
+ */
+ protected final TimelineInteger targetAssignmentEpoch;
+
+ /**
+ * The target assignment per member id.
+ */
+ protected final TimelineHashMap<String, Assignment> targetAssignment;
+
+ /**
+ * Reverse lookup map representing topic partitions with
+ * their current member assignments.
+ */
+ private final TimelineHashMap<Uuid, TimelineHashMap<Integer, String>> invertedTargetAssignment;
+
+ /**
+ * The current partition epoch maps each topic-partitions to their current epoch where
+ * the epoch is the epoch of their owners. When a member revokes a partition, it removes
+ * its epochs from this map. When a member gets a partition, it adds its epochs to this map.
+ */
+ protected final TimelineHashMap<Uuid, TimelineHashMap<Integer, Integer>> currentPartitionEpoch;
+
+ /**
+ * The metadata refresh deadline. It consists of a timestamp in milliseconds together with
+ * the group epoch at the time of setting it. The metadata refresh time is considered as a
+ * soft state (read that it is not stored in a timeline data structure). It is like this
+ * because it is not persisted to the log. The group epoch is here to ensure that the
+ * metadata refresh deadline is invalidated if the group epoch does not correspond to
+ * the current group epoch. This can happen if the metadata refresh deadline is updated
+ * after having refreshed the metadata but the write operation failed. In this case, the
+ * time is not automatically rolled back.
+ */
+ protected DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+
+ protected AbstractModernGroup(
+ SnapshotRegistry snapshotRegistry,
+ String groupId
+ ) {
+ this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
+ this.groupId = Objects.requireNonNull(groupId);
+ this.groupEpoch = new TimelineInteger(snapshotRegistry);
+ this.members = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicNames = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscriptionType = new TimelineObject<>(snapshotRegistry, HOMOGENEOUS);
+ this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
+ this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.invertedTargetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.currentPartitionEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
+ }
+
+ /**
+ * @return the group formatted as a list group response based on the committed offset.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup(long committedOffset) {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
Review Comment:
I don't think an AbstractModernGroup should contain a concrete protocol type. I suggest the protocol type comes from the concrete subclasses.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1625778608
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
Now that I see the use of `modern` in the code, I'm pretty comfortable with it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #16054:
URL: https://github.com/apache/kafka/pull/16054#issuecomment-2127858693
@dajac @omkreddy @AndrewJSchofield Please review, we can merge post 3.8 branch cut.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1616442484
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
+
+ public static class DeadlineAndEpoch {
+ static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0);
+
+ public final long deadlineMs;
+ public final int epoch;
+
+ DeadlineAndEpoch(long deadlineMs, int epoch) {
+ this.deadlineMs = deadlineMs;
+ this.epoch = epoch;
+ }
+ }
+
+ /**
+ * The snapshot registry.
+ */
+ protected final SnapshotRegistry snapshotRegistry;
+
+ /**
+ * The group id.
+ */
+ protected final String groupId;
+
+ /**
+ * The group epoch. The epoch is incremented whenever the subscriptions
+ * are updated and it will trigger the computation of a new assignment
+ * for the group.
+ */
+ protected final TimelineInteger groupEpoch;
+
+ /**
+ * The group members.
+ */
+ protected final TimelineHashMap<String, GroupMember> members;
+
+ /**
+ * The number of subscribers per topic.
+ */
+ protected final TimelineHashMap<String, Integer> subscribedTopicNames;
+
+ /**
+ * The metadata associated with each subscribed topic name.
+ */
+ protected final TimelineHashMap<String, TopicMetadata> subscribedTopicMetadata;
+
+ /**
+ * The group's subscription type.
+ * This value is set to Homogeneous by default.
+ */
+ protected final TimelineObject<SubscriptionType> subscriptionType;
+
+ /**
+ * The target assignment epoch. An assignment epoch smaller than the group epoch
+ * means that a new assignment is required. The assignment epoch is updated when
+ * a new assignment is installed.
+ */
+ protected final TimelineInteger targetAssignmentEpoch;
+
+ /**
+ * The target assignment per member id.
+ */
+ protected final TimelineHashMap<String, Assignment> targetAssignment;
+
+ /**
+ * The current partition epoch maps each topic-partitions to their current epoch where
+ * the epoch is the epoch of their owners. When a member revokes a partition, it removes
+ * its epochs from this map. When a member gets a partition, it adds its epochs to this map.
+ */
+ protected final TimelineHashMap<Uuid, TimelineHashMap<Integer, Integer>> currentPartitionEpoch;
+
+ /**
+ * The metadata refresh deadline. It consists of a timestamp in milliseconds together with
+ * the group epoch at the time of setting it. The metadata refresh time is considered as a
+ * soft state (read that it is not stored in a timeline data structure). It is like this
+ * because it is not persisted to the log. The group epoch is here to ensure that the
+ * metadata refresh deadline is invalidated if the group epoch does not correspond to
+ * the current group epoch. This can happen if the metadata refresh deadline is updated
+ * after having refreshed the metadata but the write operation failed. In this case, the
+ * time is not automatically rolled back.
+ */
+ protected DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+
+ protected AbstractGroup(
+ SnapshotRegistry snapshotRegistry,
+ String groupId
+ ) {
+ this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
+ this.groupId = Objects.requireNonNull(groupId);
+ this.groupEpoch = new TimelineInteger(snapshotRegistry);
+ this.members = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicNames = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscriptionType = new TimelineObject<>(snapshotRegistry, HOMOGENEOUS);
+ this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
+ this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.currentPartitionEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
+ }
+
+ /**
+ * @return the group formatted as a list group response based on the committed offset.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup(long committedOffset) {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(stateAsString(committedOffset))
+ .setGroupType(type().toString());
+ }
+
+ /**
+ * @return The group id.
+ */
+ @Override
+ public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * @return The group epoch.
+ */
+ public int groupEpoch() {
+ return groupEpoch.get();
+ }
+
+ /**
+ * Sets the group epoch.
+ *
+ * @param groupEpoch The new group epoch.
+ */
+ public void setGroupEpoch(int groupEpoch) {
+ this.groupEpoch.set(groupEpoch);
+ maybeUpdateGroupState();
+ }
+
+ /**
+ * @return The target assignment epoch.
+ */
+ public int assignmentEpoch() {
+ return targetAssignmentEpoch.get();
+ }
+
+ /**
+ * Sets the assignment epoch.
+ *
+ * @param targetAssignmentEpoch The new assignment epoch.
+ */
+ public void setTargetAssignmentEpoch(int targetAssignmentEpoch) {
+ this.targetAssignmentEpoch.set(targetAssignmentEpoch);
+ maybeUpdateGroupState();
+ }
+
+ /**
+ * @return A boolean indicating whether the member exists or not.
+ */
+ @Override
+ public boolean hasMember(String memberId) {
+ return members.containsKey(memberId);
+ }
+
+ /**
+ * @return The number of members.
+ */
+ @Override
+ public int numMembers() {
+ return members.size();
+ }
+
+ /**
+ * @return An immutable Map containing all the members keyed by their id.
+ */
+ public Map<String, GroupMember> members() {
+ return Collections.unmodifiableMap(members);
+ }
+
+ /**
+ * @return An immutable map containing all the subscribed topic names
+ * with the subscribers counts per topic.
+ */
+ public Map<String, Integer> subscribedTopicNames() {
+ return Collections.unmodifiableMap(subscribedTopicNames);
+ }
+
+ /**
+ * Returns true if the group is actively subscribed to the topic.
+ *
+ * @param topic The topic name.
+ *
+ * @return Whether the group is subscribed to the topic.
+ */
+ @Override
+ public boolean isSubscribedToTopic(String topic) {
+ return subscribedTopicNames.containsKey(topic);
+ }
+
+ /**
+ * @return The group's subscription type.
+ */
+ public SubscriptionType subscriptionType() {
+ return subscriptionType.get();
+ }
+
+ /**
+ * Returns the target assignment of the member.
+ *
+ * @return The ConsumerGroupMemberAssignment or an EMPTY one if it does not
+ * exist.
+ */
+ public Assignment targetAssignment(String memberId) {
+ return targetAssignment.getOrDefault(memberId, Assignment.EMPTY);
+ }
+
+ /**
+ * Updates target assignment of a member.
+ *
+ * @param memberId The member id.
+ * @param newTargetAssignment The new target assignment.
+ */
+ public void updateTargetAssignment(String memberId, Assignment newTargetAssignment) {
+ targetAssignment.put(memberId, newTargetAssignment);
+ }
+
+ /**
+ * Removes the target assignment of a member.
+ *
+ * @param memberId The member id.
+ */
+ public void removeTargetAssignment(String memberId) {
+ targetAssignment.remove(memberId);
+ }
+
+ /**
+ * @return An immutable Map containing all the target assignment keyed by member id.
+ */
+ public Map<String, Assignment> targetAssignment() {
+ return Collections.unmodifiableMap(targetAssignment);
+ }
+
+ /**
+ * Returns the current epoch of a partition or -1 if the partition
+ * does not have one.
+ *
+ * @param topicId The topic id.
+ * @param partitionId The partition id.
+ *
+ * @return The epoch or -1.
+ */
+ public int currentPartitionEpoch(
+ Uuid topicId, int partitionId
+ ) {
+ Map<Integer, Integer> partitions = currentPartitionEpoch.get(topicId);
+ if (partitions == null) {
+ return -1;
+ } else {
+ return partitions.getOrDefault(partitionId, -1);
+ }
+ }
+
+ /**
+ * @return An immutable Map of subscription metadata for
+ * each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> subscriptionMetadata() {
+ return Collections.unmodifiableMap(subscribedTopicMetadata);
+ }
+
+ /**
+ * Updates the subscription metadata. This replaces the previous one.
+ *
+ * @param subscriptionMetadata The new subscription metadata.
+ */
+ public void setSubscriptionMetadata(
+ Map<String, TopicMetadata> subscriptionMetadata
+ ) {
+ this.subscribedTopicMetadata.clear();
+ this.subscribedTopicMetadata.putAll(subscriptionMetadata);
+ }
+
+ /**
+ * Computes the subscription metadata based on the current subscription info.
+ *
+ * @param subscribedTopicNames Map of topic names to the number of subscribers.
+ * @param topicsImage The current metadata for all available topics.
+ * @param clusterImage The current metadata for the Kafka cluster.
+ *
+ * @return An immutable map of subscription metadata for each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> computeSubscriptionMetadata(
+ Map<String, Integer> subscribedTopicNames,
+ TopicsImage topicsImage,
+ ClusterImage clusterImage
+ ) {
+ // Create the topic metadata for each subscribed topic.
+ Map<String, TopicMetadata> newSubscriptionMetadata = new HashMap<>(subscribedTopicNames.size());
+
+ subscribedTopicNames.forEach((topicName, count) -> {
+ TopicImage topicImage = topicsImage.getTopic(topicName);
+ if (topicImage != null) {
+ Map<Integer, Set<String>> partitionRacks = new HashMap<>();
+ topicImage.partitions().forEach((partition, partitionRegistration) -> {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ Optional<String> rackOptional = clusterImage.broker(replica).rack();
+ // Only add the rack if it is available for the broker/replica.
+ rackOptional.ifPresent(racks::add);
+ }
+ // If rack information is unavailable for all replicas of this partition,
+ // no corresponding entry will be stored for it in the map.
+ if (!racks.isEmpty())
+ partitionRacks.put(partition, racks);
+ });
+
+ newSubscriptionMetadata.put(topicName, new TopicMetadata(
+ topicImage.id(),
+ topicImage.name(),
+ topicImage.partitions().size(),
+ partitionRacks)
+ );
+ }
+ });
+
+ return Collections.unmodifiableMap(newSubscriptionMetadata);
+ }
+
+ /**
+ * Updates the metadata refresh deadline.
+ *
+ * @param deadlineMs The deadline in milliseconds.
+ * @param groupEpoch The associated group epoch.
+ */
+ public void setMetadataRefreshDeadline(
+ long deadlineMs,
+ int groupEpoch
+ ) {
+ this.metadataRefreshDeadline = new DeadlineAndEpoch(deadlineMs, groupEpoch);
+ }
+
+ /**
+ * Requests a metadata refresh.
+ */
+ public void requestMetadataRefresh() {
+ this.metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+ }
+
+ /**
+ * Checks if a metadata refresh is required. A refresh is required in two cases:
+ * 1) The deadline is smaller or equal to the current time;
+ * 2) The group epoch associated with the deadline is larger than
+ * the current group epoch. This means that the operations which updated
+ * the deadline failed.
+ *
+ * @param currentTimeMs The current time in milliseconds.
+ * @return A boolean indicating whether a refresh is required or not.
+ */
+ public boolean hasMetadataExpired(long currentTimeMs) {
+ return currentTimeMs >= metadataRefreshDeadline.deadlineMs || groupEpoch() < metadataRefreshDeadline.epoch;
+ }
+
+ /**
+ * @return The metadata refresh deadline.
+ */
+ public DeadlineAndEpoch metadataRefreshDeadline() {
+ return metadataRefreshDeadline;
+ }
+
+ /**
+ * Validates the OffsetCommit request.
+ *
+ * @param memberId The member id.
+ * @param groupInstanceId The group instance id.
+ * @param memberEpoch The member epoch.
+ * @param isTransactional Whether the offset commit is transactional or not. It has no
+ * impact when a consumer group is used.
+ */
+ @Override
+ public void validateOffsetCommit(
+ String memberId,
+ String groupInstanceId,
+ int memberEpoch,
+ boolean isTransactional
+ ) throws UnknownMemberIdException, StaleMemberEpochException {
+ // When the member epoch is -1, the request comes from either the admin client
+ // or a consumer which does not use the group management facility. In this case,
+ // the request can commit offsets if the group is empty.
+ if (memberEpoch < 0 && members().isEmpty()) return;
+
+ final GroupMember member = getOrMaybeCreateMember(memberId, false);
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ }
+
+ /**
+ * Validates the OffsetFetch request.
+ *
+ * @param memberId The member id for consumer groups.
+ * @param memberEpoch The member epoch for consumer groups.
+ * @param lastCommittedOffset The last committed offsets in the timeline.
+ */
+ @Override
+ public void validateOffsetFetch(
+ String memberId,
+ int memberEpoch,
+ long lastCommittedOffset
+ ) throws UnknownMemberIdException, StaleMemberEpochException {
+ // When the member id is null and the member epoch is -1, the request either comes
+ // from the admin client or from a client which does not provide them. In this case,
+ // the fetch request is accepted.
+ if (memberId == null && memberEpoch < 0) return;
+
+ final GroupMember member = members.get(memberId, lastCommittedOffset);
+ if (member == null) {
+ throw new UnknownMemberIdException(String.format("Member %s is not a member of group %s.",
+ memberId, groupId));
+ }
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ }
+
+ /**
+ * Validates the OffsetDelete request.
+ */
+ @Override
+ public void validateOffsetDelete() {
+ // Do nothing.
+ }
+
+ /**
+ * See {@link org.apache.kafka.coordinator.group.OffsetExpirationCondition}
+ *
+ * @return The offset expiration condition for the group or Empty if no such condition exists.
+ */
+ @Override
+ public Optional<OffsetExpirationCondition> offsetExpirationCondition() {
+ return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata -> offsetAndMetadata.commitTimestampMs));
+ }
+
+ /**
+ * Throws a StaleMemberEpochException if the received member epoch does not match
+ * the expected member epoch.
+ */
+ private void validateMemberEpoch(
+ int receivedMemberEpoch,
+ int expectedMemberEpoch
+ ) throws StaleMemberEpochException {
+ if (receivedMemberEpoch != expectedMemberEpoch) {
+ throw new StaleMemberEpochException(String.format("The received member epoch %d does not match "
+ + "the expected member epoch %d.", receivedMemberEpoch, expectedMemberEpoch));
+ }
+ }
+
+ /**
+ * Updates the subscribed topic names count.
+ * The subscription type is updated as a consequence.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ protected void maybeUpdateSubscribedTopicNamesAndGroupSubscriptionType(
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ maybeUpdateSubscribedTopicNames(subscribedTopicNames, oldMember, newMember);
+ subscriptionType.set(subscriptionType(subscribedTopicNames, members.size()));
+ }
+
+ /**
+ * Updates the subscription count.
+ *
+ * @param subscribedTopicCount The map to update.
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ private static void maybeUpdateSubscribedTopicNames(
+ Map<String, Integer> subscribedTopicCount,
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ if (oldMember != null) {
+ oldMember.subscribedTopicNames().forEach(topicName ->
+ subscribedTopicCount.compute(topicName, ConsumerGroup::decValue)
+ );
+ }
+
+ if (newMember != null) {
+ newMember.subscribedTopicNames().forEach(topicName ->
+ subscribedTopicCount.compute(topicName, ConsumerGroup::incValue)
+ );
+ }
+ }
+
+ /**
+ * Updates the subscription count.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ *
+ * @return Copy of the map of topics to the count of number of subscribers.
+ */
+ public Map<String, Integer> computeSubscribedTopicNames(
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ Map<String, Integer> subscribedTopicNames = new HashMap<>(this.subscribedTopicNames);
+ maybeUpdateSubscribedTopicNames(
+ subscribedTopicNames,
+ oldMember,
+ newMember
+ );
+ return subscribedTopicNames;
+ }
+
+ /**
+ * Compute the subscription type of the consumer group.
+ *
+ * @param subscribedTopicNames A map of topic names to the count of members subscribed to each topic.
+ *
+ * @return {@link SubscriptionType#HOMOGENEOUS} if all members are subscribed to exactly the same topics;
+ * otherwise, {@link SubscriptionType#HETEROGENEOUS}.
+ */
+ public static SubscriptionType subscriptionType(
+ Map<String, Integer> subscribedTopicNames,
+ int numberOfMembers
+ ) {
+ if (subscribedTopicNames.isEmpty()) {
+ return HOMOGENEOUS;
+ }
+
+ for (int subscriberCount : subscribedTopicNames.values()) {
+ if (subscriberCount != numberOfMembers) {
+ return HETEROGENEOUS;
+ }
+ }
+ return HOMOGENEOUS;
+ }
+
+ /**
+ * Decrements value by 1; returns null when reaching zero. This helper is
+ * meant to be used with Map#compute.
+ */
+ protected static Integer decValue(String key, Integer value) {
+ if (value == null) return null;
+ return value == 1 ? null : value - 1;
+ }
+
+ /**
+ * Increments value by 1; This helper is meant to be used with Map#compute.
+ */
+ protected static Integer incValue(String key, Integer value) {
+ return value == null ? 1 : value + 1;
+ }
+
+ /**
+ * Gets or creates a member.
+ *
+ * @param memberId The member id.
+ * @param createIfNotExists Booleans indicating whether the member must be
+ * created if it does not exist.
+ *
+ * @return A ConsumerGroupMember.
+ */
+ public abstract GroupMember getOrMaybeCreateMember(String memberId, boolean createIfNotExists);
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613579324
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
The protocol behind the groups seems to be the main distinction as I see it (meaning the base protocol). Classic groups as part of the classic protocol, vs share/consumer/streams.. that all build on top of the new protocol. Other than that, the groups are quite different I would say, so hard to find a common feature. Seeing it like this, the non-classic seems like a sensible umbrella term (given that classic is the term we are standardizing for the old group and old protocol, so convenient hint to the protocol too).
No matter the name, I would strongly suggest we add a the java doc for this class btw.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613686172
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
I also wonder if streams will really use this class because the assigned resources will be totally different, I suppose.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1616442554
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -128,10 +129,10 @@ public void updateMemberSubscription(
Optional<String> instanceId,
Optional<String> rackId
) {
- ConsumerGroupMember existingMember = members.get(memberId);
+ GroupMember existingMember = members.get(memberId);
ConsumerGroupMember.Builder builder;
if (existingMember != null) {
- builder = new ConsumerGroupMember.Builder(existingMember);
+ builder = new ConsumerGroupMember.Builder((ConsumerGroupMember) existingMember);
Review Comment:
Agree, resolved.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1617009938
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
+
+ public static class DeadlineAndEpoch {
+ static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0);
+
+ public final long deadlineMs;
+ public final int epoch;
+
+ DeadlineAndEpoch(long deadlineMs, int epoch) {
+ this.deadlineMs = deadlineMs;
+ this.epoch = epoch;
+ }
+ }
+
+ /**
+ * The snapshot registry.
+ */
+ protected final SnapshotRegistry snapshotRegistry;
+
+ /**
+ * The group id.
+ */
+ protected final String groupId;
+
+ /**
+ * The group epoch. The epoch is incremented whenever the subscriptions
+ * are updated and it will trigger the computation of a new assignment
+ * for the group.
+ */
+ protected final TimelineInteger groupEpoch;
+
+ /**
+ * The group members.
+ */
+ protected final TimelineHashMap<String, GroupMember> members;
+
+ /**
+ * The number of subscribers per topic.
+ */
+ protected final TimelineHashMap<String, Integer> subscribedTopicNames;
+
+ /**
+ * The metadata associated with each subscribed topic name.
+ */
+ protected final TimelineHashMap<String, TopicMetadata> subscribedTopicMetadata;
+
+ /**
+ * The group's subscription type.
+ * This value is set to Homogeneous by default.
+ */
+ protected final TimelineObject<SubscriptionType> subscriptionType;
+
+ /**
+ * The target assignment epoch. An assignment epoch smaller than the group epoch
+ * means that a new assignment is required. The assignment epoch is updated when
+ * a new assignment is installed.
+ */
+ protected final TimelineInteger targetAssignmentEpoch;
+
+ /**
+ * The target assignment per member id.
+ */
+ protected final TimelineHashMap<String, Assignment> targetAssignment;
+
+ /**
+ * The current partition epoch maps each topic-partitions to their current epoch where
+ * the epoch is the epoch of their owners. When a member revokes a partition, it removes
+ * its epochs from this map. When a member gets a partition, it adds its epochs to this map.
+ */
+ protected final TimelineHashMap<Uuid, TimelineHashMap<Integer, Integer>> currentPartitionEpoch;
+
+ /**
+ * The metadata refresh deadline. It consists of a timestamp in milliseconds together with
+ * the group epoch at the time of setting it. The metadata refresh time is considered as a
+ * soft state (read that it is not stored in a timeline data structure). It is like this
+ * because it is not persisted to the log. The group epoch is here to ensure that the
+ * metadata refresh deadline is invalidated if the group epoch does not correspond to
+ * the current group epoch. This can happen if the metadata refresh deadline is updated
+ * after having refreshed the metadata but the write operation failed. In this case, the
+ * time is not automatically rolled back.
+ */
+ protected DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+
+ protected AbstractGroup(
+ SnapshotRegistry snapshotRegistry,
+ String groupId
+ ) {
+ this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
+ this.groupId = Objects.requireNonNull(groupId);
+ this.groupEpoch = new TimelineInteger(snapshotRegistry);
+ this.members = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicNames = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscriptionType = new TimelineObject<>(snapshotRegistry, HOMOGENEOUS);
+ this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
+ this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.currentPartitionEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
+ }
+
+ /**
+ * @return the group formatted as a list group response based on the committed offset.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup(long committedOffset) {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(stateAsString(committedOffset))
+ .setGroupType(type().toString());
+ }
+
+ /**
+ * @return The group id.
+ */
+ @Override
+ public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * @return The group epoch.
+ */
+ public int groupEpoch() {
+ return groupEpoch.get();
+ }
+
+ /**
+ * Sets the group epoch.
+ *
+ * @param groupEpoch The new group epoch.
+ */
+ public void setGroupEpoch(int groupEpoch) {
+ this.groupEpoch.set(groupEpoch);
+ maybeUpdateGroupState();
+ }
+
+ /**
+ * @return The target assignment epoch.
+ */
+ public int assignmentEpoch() {
+ return targetAssignmentEpoch.get();
+ }
+
+ /**
+ * Sets the assignment epoch.
+ *
+ * @param targetAssignmentEpoch The new assignment epoch.
+ */
+ public void setTargetAssignmentEpoch(int targetAssignmentEpoch) {
+ this.targetAssignmentEpoch.set(targetAssignmentEpoch);
+ maybeUpdateGroupState();
+ }
+
+ /**
+ * @return A boolean indicating whether the member exists or not.
+ */
+ @Override
+ public boolean hasMember(String memberId) {
+ return members.containsKey(memberId);
+ }
+
+ /**
+ * @return The number of members.
+ */
+ @Override
+ public int numMembers() {
+ return members.size();
+ }
+
+ /**
+ * @return An immutable Map containing all the members keyed by their id.
+ */
+ public Map<String, GroupMember> members() {
+ return Collections.unmodifiableMap(members);
+ }
+
+ /**
+ * @return An immutable map containing all the subscribed topic names
+ * with the subscribers counts per topic.
+ */
+ public Map<String, Integer> subscribedTopicNames() {
+ return Collections.unmodifiableMap(subscribedTopicNames);
+ }
+
+ /**
+ * Returns true if the group is actively subscribed to the topic.
+ *
+ * @param topic The topic name.
+ *
+ * @return Whether the group is subscribed to the topic.
+ */
+ @Override
+ public boolean isSubscribedToTopic(String topic) {
+ return subscribedTopicNames.containsKey(topic);
+ }
+
+ /**
+ * @return The group's subscription type.
+ */
+ public SubscriptionType subscriptionType() {
+ return subscriptionType.get();
+ }
+
+ /**
+ * Returns the target assignment of the member.
+ *
+ * @return The ConsumerGroupMemberAssignment or an EMPTY one if it does not
+ * exist.
+ */
+ public Assignment targetAssignment(String memberId) {
+ return targetAssignment.getOrDefault(memberId, Assignment.EMPTY);
+ }
+
+ /**
+ * Updates target assignment of a member.
+ *
+ * @param memberId The member id.
+ * @param newTargetAssignment The new target assignment.
+ */
+ public void updateTargetAssignment(String memberId, Assignment newTargetAssignment) {
+ targetAssignment.put(memberId, newTargetAssignment);
+ }
+
+ /**
+ * Removes the target assignment of a member.
+ *
+ * @param memberId The member id.
+ */
+ public void removeTargetAssignment(String memberId) {
+ targetAssignment.remove(memberId);
+ }
+
+ /**
+ * @return An immutable Map containing all the target assignment keyed by member id.
+ */
+ public Map<String, Assignment> targetAssignment() {
+ return Collections.unmodifiableMap(targetAssignment);
+ }
+
+ /**
+ * Returns the current epoch of a partition or -1 if the partition
+ * does not have one.
+ *
+ * @param topicId The topic id.
+ * @param partitionId The partition id.
+ *
+ * @return The epoch or -1.
+ */
+ public int currentPartitionEpoch(
+ Uuid topicId, int partitionId
+ ) {
+ Map<Integer, Integer> partitions = currentPartitionEpoch.get(topicId);
+ if (partitions == null) {
+ return -1;
+ } else {
+ return partitions.getOrDefault(partitionId, -1);
+ }
+ }
+
+ /**
+ * @return An immutable Map of subscription metadata for
+ * each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> subscriptionMetadata() {
+ return Collections.unmodifiableMap(subscribedTopicMetadata);
+ }
+
+ /**
+ * Updates the subscription metadata. This replaces the previous one.
+ *
+ * @param subscriptionMetadata The new subscription metadata.
+ */
+ public void setSubscriptionMetadata(
+ Map<String, TopicMetadata> subscriptionMetadata
+ ) {
+ this.subscribedTopicMetadata.clear();
+ this.subscribedTopicMetadata.putAll(subscriptionMetadata);
+ }
+
+ /**
+ * Computes the subscription metadata based on the current subscription info.
+ *
+ * @param subscribedTopicNames Map of topic names to the number of subscribers.
+ * @param topicsImage The current metadata for all available topics.
+ * @param clusterImage The current metadata for the Kafka cluster.
+ *
+ * @return An immutable map of subscription metadata for each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> computeSubscriptionMetadata(
+ Map<String, Integer> subscribedTopicNames,
+ TopicsImage topicsImage,
+ ClusterImage clusterImage
+ ) {
+ // Create the topic metadata for each subscribed topic.
+ Map<String, TopicMetadata> newSubscriptionMetadata = new HashMap<>(subscribedTopicNames.size());
+
+ subscribedTopicNames.forEach((topicName, count) -> {
+ TopicImage topicImage = topicsImage.getTopic(topicName);
+ if (topicImage != null) {
+ Map<Integer, Set<String>> partitionRacks = new HashMap<>();
+ topicImage.partitions().forEach((partition, partitionRegistration) -> {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ Optional<String> rackOptional = clusterImage.broker(replica).rack();
+ // Only add the rack if it is available for the broker/replica.
+ rackOptional.ifPresent(racks::add);
+ }
+ // If rack information is unavailable for all replicas of this partition,
+ // no corresponding entry will be stored for it in the map.
+ if (!racks.isEmpty())
+ partitionRacks.put(partition, racks);
+ });
+
+ newSubscriptionMetadata.put(topicName, new TopicMetadata(
+ topicImage.id(),
+ topicImage.name(),
+ topicImage.partitions().size(),
+ partitionRacks)
+ );
+ }
+ });
+
+ return Collections.unmodifiableMap(newSubscriptionMetadata);
+ }
+
+ /**
+ * Updates the metadata refresh deadline.
+ *
+ * @param deadlineMs The deadline in milliseconds.
+ * @param groupEpoch The associated group epoch.
+ */
+ public void setMetadataRefreshDeadline(
+ long deadlineMs,
+ int groupEpoch
+ ) {
+ this.metadataRefreshDeadline = new DeadlineAndEpoch(deadlineMs, groupEpoch);
+ }
+
+ /**
+ * Requests a metadata refresh.
+ */
+ public void requestMetadataRefresh() {
+ this.metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+ }
+
+ /**
+ * Checks if a metadata refresh is required. A refresh is required in two cases:
+ * 1) The deadline is smaller or equal to the current time;
+ * 2) The group epoch associated with the deadline is larger than
+ * the current group epoch. This means that the operations which updated
+ * the deadline failed.
+ *
+ * @param currentTimeMs The current time in milliseconds.
+ * @return A boolean indicating whether a refresh is required or not.
+ */
+ public boolean hasMetadataExpired(long currentTimeMs) {
+ return currentTimeMs >= metadataRefreshDeadline.deadlineMs || groupEpoch() < metadataRefreshDeadline.epoch;
+ }
+
+ /**
+ * @return The metadata refresh deadline.
+ */
+ public DeadlineAndEpoch metadataRefreshDeadline() {
+ return metadataRefreshDeadline;
+ }
+
+ /**
+ * Validates the OffsetCommit request.
+ *
+ * @param memberId The member id.
+ * @param groupInstanceId The group instance id.
+ * @param memberEpoch The member epoch.
+ * @param isTransactional Whether the offset commit is transactional or not. It has no
+ * impact when a consumer group is used.
+ */
+ @Override
+ public void validateOffsetCommit(
+ String memberId,
+ String groupInstanceId,
+ int memberEpoch,
+ boolean isTransactional
+ ) throws UnknownMemberIdException, StaleMemberEpochException {
+ // When the member epoch is -1, the request comes from either the admin client
+ // or a consumer which does not use the group management facility. In this case,
+ // the request can commit offsets if the group is empty.
+ if (memberEpoch < 0 && members().isEmpty()) return;
+
+ final GroupMember member = getOrMaybeCreateMember(memberId, false);
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ }
+
+ /**
+ * Validates the OffsetFetch request.
+ *
+ * @param memberId The member id for consumer groups.
+ * @param memberEpoch The member epoch for consumer groups.
+ * @param lastCommittedOffset The last committed offsets in the timeline.
+ */
+ @Override
+ public void validateOffsetFetch(
+ String memberId,
+ int memberEpoch,
+ long lastCommittedOffset
+ ) throws UnknownMemberIdException, StaleMemberEpochException {
+ // When the member id is null and the member epoch is -1, the request either comes
+ // from the admin client or from a client which does not provide them. In this case,
+ // the fetch request is accepted.
+ if (memberId == null && memberEpoch < 0) return;
+
+ final GroupMember member = members.get(memberId, lastCommittedOffset);
+ if (member == null) {
+ throw new UnknownMemberIdException(String.format("Member %s is not a member of group %s.",
+ memberId, groupId));
+ }
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ }
+
+ /**
+ * Validates the OffsetDelete request.
+ */
+ @Override
+ public void validateOffsetDelete() {
+ // Do nothing.
+ }
+
+ /**
+ * See {@link org.apache.kafka.coordinator.group.OffsetExpirationCondition}
+ *
+ * @return The offset expiration condition for the group or Empty if no such condition exists.
+ */
+ @Override
+ public Optional<OffsetExpirationCondition> offsetExpirationCondition() {
+ return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata -> offsetAndMetadata.commitTimestampMs));
+ }
+
+ /**
+ * Throws a StaleMemberEpochException if the received member epoch does not match
+ * the expected member epoch.
+ */
+ private void validateMemberEpoch(
+ int receivedMemberEpoch,
+ int expectedMemberEpoch
+ ) throws StaleMemberEpochException {
+ if (receivedMemberEpoch != expectedMemberEpoch) {
+ throw new StaleMemberEpochException(String.format("The received member epoch %d does not match "
+ + "the expected member epoch %d.", receivedMemberEpoch, expectedMemberEpoch));
+ }
+ }
+
+ /**
+ * Updates the subscribed topic names count.
+ * The subscription type is updated as a consequence.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ protected void maybeUpdateSubscribedTopicNamesAndGroupSubscriptionType(
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ maybeUpdateSubscribedTopicNames(subscribedTopicNames, oldMember, newMember);
+ subscriptionType.set(subscriptionType(subscribedTopicNames, members.size()));
+ }
+
+ /**
+ * Updates the subscription count.
+ *
+ * @param subscribedTopicCount The map to update.
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ private static void maybeUpdateSubscribedTopicNames(
+ Map<String, Integer> subscribedTopicCount,
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ if (oldMember != null) {
+ oldMember.subscribedTopicNames().forEach(topicName ->
+ subscribedTopicCount.compute(topicName, ConsumerGroup::decValue)
+ );
+ }
+
+ if (newMember != null) {
+ newMember.subscribedTopicNames().forEach(topicName ->
+ subscribedTopicCount.compute(topicName, ConsumerGroup::incValue)
+ );
+ }
+ }
+
+ /**
+ * Updates the subscription count.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ *
+ * @return Copy of the map of topics to the count of number of subscribers.
+ */
+ public Map<String, Integer> computeSubscribedTopicNames(
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ Map<String, Integer> subscribedTopicNames = new HashMap<>(this.subscribedTopicNames);
+ maybeUpdateSubscribedTopicNames(
+ subscribedTopicNames,
+ oldMember,
+ newMember
+ );
+ return subscribedTopicNames;
+ }
+
+ /**
+ * Compute the subscription type of the consumer group.
+ *
+ * @param subscribedTopicNames A map of topic names to the count of members subscribed to each topic.
+ *
+ * @return {@link SubscriptionType#HOMOGENEOUS} if all members are subscribed to exactly the same topics;
+ * otherwise, {@link SubscriptionType#HETEROGENEOUS}.
+ */
+ public static SubscriptionType subscriptionType(
+ Map<String, Integer> subscribedTopicNames,
+ int numberOfMembers
+ ) {
+ if (subscribedTopicNames.isEmpty()) {
+ return HOMOGENEOUS;
+ }
+
+ for (int subscriberCount : subscribedTopicNames.values()) {
+ if (subscriberCount != numberOfMembers) {
+ return HETEROGENEOUS;
+ }
+ }
+ return HOMOGENEOUS;
+ }
+
+ /**
+ * Decrements value by 1; returns null when reaching zero. This helper is
+ * meant to be used with Map#compute.
+ */
+ protected static Integer decValue(String key, Integer value) {
+ if (value == null) return null;
+ return value == 1 ? null : value - 1;
+ }
+
+ /**
+ * Increments value by 1; This helper is meant to be used with Map#compute.
+ */
+ protected static Integer incValue(String key, Integer value) {
+ return value == null ? 1 : value + 1;
+ }
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on PR #16054:
URL: https://github.com/apache/kafka/pull/16054#issuecomment-2141730615
@dajac @lianetm @AndrewJSchofield Can I please get a review again, have addressed the comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1617014505
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
I have renamed `AbstractGroup` to `AbstractModernGroup` and `GroupMember` to `ModernGroupMember`. Also I have created a common package named `modern` as suggested above.
This PR mostly does the refactor around adding abstract classes and in the next PR I will move `consumer` package under `modern`, also will move couple of shared classes from `consumer` to `modern` as well. Just don't to increase more diffs in the current PR with additional refactoring.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "apoorvmittal10 (via GitHub)" <gi...@apache.org>.
apoorvmittal10 commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1626216500
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/modern/AbstractModernGroup.java:
##########
@@ -0,0 +1,713 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group.modern;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.Group;
+import org.apache.kafka.coordinator.group.OffsetExpirationCondition;
+import org.apache.kafka.coordinator.group.OffsetExpirationConditionImpl;
+import org.apache.kafka.coordinator.group.Utils;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+/**
+ * The abstract group provides definitions for the consumer and share group.
+ */
+public abstract class AbstractModernGroup<T extends ModernGroupMember> implements Group {
+
+ public static class DeadlineAndEpoch {
+ static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0);
+
+ public final long deadlineMs;
+ public final int epoch;
+
+ DeadlineAndEpoch(long deadlineMs, int epoch) {
+ this.deadlineMs = deadlineMs;
+ this.epoch = epoch;
+ }
+ }
+
+ /**
+ * The snapshot registry.
+ */
+ protected final SnapshotRegistry snapshotRegistry;
+
+ /**
+ * The group id.
+ */
+ protected final String groupId;
+
+ /**
+ * The group epoch. The epoch is incremented whenever the subscriptions
+ * are updated and it will trigger the computation of a new assignment
+ * for the group.
+ */
+ protected final TimelineInteger groupEpoch;
+
+ /**
+ * The group members.
+ */
+ protected final TimelineHashMap<String, T> members;
+
+ /**
+ * The number of subscribers per topic.
+ */
+ protected final TimelineHashMap<String, Integer> subscribedTopicNames;
+
+ /**
+ * The metadata associated with each subscribed topic name.
+ */
+ protected final TimelineHashMap<String, TopicMetadata> subscribedTopicMetadata;
+
+ /**
+ * The group's subscription type.
+ * This value is set to Homogeneous by default.
+ */
+ protected final TimelineObject<SubscriptionType> subscriptionType;
+
+ /**
+ * The target assignment epoch. An assignment epoch smaller than the group epoch
+ * means that a new assignment is required. The assignment epoch is updated when
+ * a new assignment is installed.
+ */
+ protected final TimelineInteger targetAssignmentEpoch;
+
+ /**
+ * The target assignment per member id.
+ */
+ protected final TimelineHashMap<String, Assignment> targetAssignment;
+
+ /**
+ * Reverse lookup map representing topic partitions with
+ * their current member assignments.
+ */
+ private final TimelineHashMap<Uuid, TimelineHashMap<Integer, String>> invertedTargetAssignment;
+
+ /**
+ * The current partition epoch maps each topic-partitions to their current epoch where
+ * the epoch is the epoch of their owners. When a member revokes a partition, it removes
+ * its epochs from this map. When a member gets a partition, it adds its epochs to this map.
+ */
+ protected final TimelineHashMap<Uuid, TimelineHashMap<Integer, Integer>> currentPartitionEpoch;
+
+ /**
+ * The metadata refresh deadline. It consists of a timestamp in milliseconds together with
+ * the group epoch at the time of setting it. The metadata refresh time is considered as a
+ * soft state (read that it is not stored in a timeline data structure). It is like this
+ * because it is not persisted to the log. The group epoch is here to ensure that the
+ * metadata refresh deadline is invalidated if the group epoch does not correspond to
+ * the current group epoch. This can happen if the metadata refresh deadline is updated
+ * after having refreshed the metadata but the write operation failed. In this case, the
+ * time is not automatically rolled back.
+ */
+ protected DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+
+ protected AbstractModernGroup(
+ SnapshotRegistry snapshotRegistry,
+ String groupId
+ ) {
+ this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
+ this.groupId = Objects.requireNonNull(groupId);
+ this.groupEpoch = new TimelineInteger(snapshotRegistry);
+ this.members = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicNames = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscriptionType = new TimelineObject<>(snapshotRegistry, HOMOGENEOUS);
+ this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
+ this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.invertedTargetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.currentPartitionEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
+ }
+
+ /**
+ * @return the group formatted as a list group response based on the committed offset.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup(long committedOffset) {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
Review Comment:
I have exposed a method to fetch the right protocol and re-used it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613579324
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
The protocol behind the groups seems to be the main distinction as I see it (meaning the base protocol). Classic groups as part of the classic protocol, vs share/consumer/streams.. that all build on top of the new protocol). Other than that, the groups are quite different I would say, so hard to find a common feature. Seeing it like this, the non-classic seems like a sensible umbrella term (given that classic is the term we are standardizing for the old group and old protocol, so convenient hint to the protocol too).
No matter the name, I would strongly suggest we add a the java doc for this class btw.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613325546
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
I think your point is a good one, but listing the two current subclasses seems problematic when we add streams later on. I wonder if there's some other common feature that we could put in the name. I suppose they're non-classic protocol groups. I hesitate to suggest AbstractModernGroup, but I expect you see what I mean.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613579324
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
The protocol behind the groups seems to be the main distinction as I see it (meaning the base protocol). Classic groups as part of the classic protocol, vs share/consumer/streams.. that all build on top of the new protocol. Other than that, the groups are quite different I would say, so hard to find a common feature. Seeing it like this, the non-classic seems like a sensible umbrella term (given that coincidently, "classic" is the term we are standardizing for the old group and old protocol, so convenient hint to the protocol too).
No matter the name, I would strongly suggest we add a the java doc for this class btw.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #16054:
URL: https://github.com/apache/kafka/pull/16054#discussion_r1613013960
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
Review Comment:
Should it be in common too?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
Review Comment:
The name is not ideal because it gives the impression that all group types extends this. I would rather use something like `AbstractConsumerAndShareGroup`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
+
+ public static class DeadlineAndEpoch {
+ static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0);
+
+ public final long deadlineMs;
+ public final int epoch;
+
+ DeadlineAndEpoch(long deadlineMs, int epoch) {
+ this.deadlineMs = deadlineMs;
+ this.epoch = epoch;
+ }
+ }
+
+ /**
+ * The snapshot registry.
+ */
+ protected final SnapshotRegistry snapshotRegistry;
+
+ /**
+ * The group id.
+ */
+ protected final String groupId;
+
+ /**
+ * The group epoch. The epoch is incremented whenever the subscriptions
+ * are updated and it will trigger the computation of a new assignment
+ * for the group.
+ */
+ protected final TimelineInteger groupEpoch;
+
+ /**
+ * The group members.
+ */
+ protected final TimelineHashMap<String, GroupMember> members;
+
+ /**
+ * The number of subscribers per topic.
+ */
+ protected final TimelineHashMap<String, Integer> subscribedTopicNames;
+
+ /**
+ * The metadata associated with each subscribed topic name.
+ */
+ protected final TimelineHashMap<String, TopicMetadata> subscribedTopicMetadata;
+
+ /**
+ * The group's subscription type.
+ * This value is set to Homogeneous by default.
+ */
+ protected final TimelineObject<SubscriptionType> subscriptionType;
+
+ /**
+ * The target assignment epoch. An assignment epoch smaller than the group epoch
+ * means that a new assignment is required. The assignment epoch is updated when
+ * a new assignment is installed.
+ */
+ protected final TimelineInteger targetAssignmentEpoch;
+
+ /**
+ * The target assignment per member id.
+ */
+ protected final TimelineHashMap<String, Assignment> targetAssignment;
+
+ /**
+ * The current partition epoch maps each topic-partitions to their current epoch where
+ * the epoch is the epoch of their owners. When a member revokes a partition, it removes
+ * its epochs from this map. When a member gets a partition, it adds its epochs to this map.
+ */
+ protected final TimelineHashMap<Uuid, TimelineHashMap<Integer, Integer>> currentPartitionEpoch;
+
+ /**
+ * The metadata refresh deadline. It consists of a timestamp in milliseconds together with
+ * the group epoch at the time of setting it. The metadata refresh time is considered as a
+ * soft state (read that it is not stored in a timeline data structure). It is like this
+ * because it is not persisted to the log. The group epoch is here to ensure that the
+ * metadata refresh deadline is invalidated if the group epoch does not correspond to
+ * the current group epoch. This can happen if the metadata refresh deadline is updated
+ * after having refreshed the metadata but the write operation failed. In this case, the
+ * time is not automatically rolled back.
+ */
+ protected DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+
+ protected AbstractGroup(
+ SnapshotRegistry snapshotRegistry,
+ String groupId
+ ) {
+ this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
+ this.groupId = Objects.requireNonNull(groupId);
+ this.groupEpoch = new TimelineInteger(snapshotRegistry);
+ this.members = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicNames = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscriptionType = new TimelineObject<>(snapshotRegistry, HOMOGENEOUS);
+ this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
+ this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.currentPartitionEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
+ }
+
+ /**
+ * @return the group formatted as a list group response based on the committed offset.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup(long committedOffset) {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(stateAsString(committedOffset))
+ .setGroupType(type().toString());
+ }
+
+ /**
+ * @return The group id.
+ */
+ @Override
+ public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * @return The group epoch.
+ */
+ public int groupEpoch() {
+ return groupEpoch.get();
+ }
+
+ /**
+ * Sets the group epoch.
+ *
+ * @param groupEpoch The new group epoch.
+ */
+ public void setGroupEpoch(int groupEpoch) {
+ this.groupEpoch.set(groupEpoch);
+ maybeUpdateGroupState();
+ }
+
+ /**
+ * @return The target assignment epoch.
+ */
+ public int assignmentEpoch() {
+ return targetAssignmentEpoch.get();
+ }
+
+ /**
+ * Sets the assignment epoch.
+ *
+ * @param targetAssignmentEpoch The new assignment epoch.
+ */
+ public void setTargetAssignmentEpoch(int targetAssignmentEpoch) {
+ this.targetAssignmentEpoch.set(targetAssignmentEpoch);
+ maybeUpdateGroupState();
+ }
+
+ /**
+ * @return A boolean indicating whether the member exists or not.
+ */
+ @Override
+ public boolean hasMember(String memberId) {
+ return members.containsKey(memberId);
+ }
+
+ /**
+ * @return The number of members.
+ */
+ @Override
+ public int numMembers() {
+ return members.size();
+ }
+
+ /**
+ * @return An immutable Map containing all the members keyed by their id.
+ */
+ public Map<String, GroupMember> members() {
+ return Collections.unmodifiableMap(members);
+ }
+
+ /**
+ * @return An immutable map containing all the subscribed topic names
+ * with the subscribers counts per topic.
+ */
+ public Map<String, Integer> subscribedTopicNames() {
+ return Collections.unmodifiableMap(subscribedTopicNames);
+ }
+
+ /**
+ * Returns true if the group is actively subscribed to the topic.
+ *
+ * @param topic The topic name.
+ *
+ * @return Whether the group is subscribed to the topic.
+ */
+ @Override
+ public boolean isSubscribedToTopic(String topic) {
+ return subscribedTopicNames.containsKey(topic);
+ }
+
+ /**
+ * @return The group's subscription type.
+ */
+ public SubscriptionType subscriptionType() {
+ return subscriptionType.get();
+ }
+
+ /**
+ * Returns the target assignment of the member.
+ *
+ * @return The ConsumerGroupMemberAssignment or an EMPTY one if it does not
+ * exist.
+ */
+ public Assignment targetAssignment(String memberId) {
+ return targetAssignment.getOrDefault(memberId, Assignment.EMPTY);
+ }
+
+ /**
+ * Updates target assignment of a member.
+ *
+ * @param memberId The member id.
+ * @param newTargetAssignment The new target assignment.
+ */
+ public void updateTargetAssignment(String memberId, Assignment newTargetAssignment) {
+ targetAssignment.put(memberId, newTargetAssignment);
+ }
+
+ /**
+ * Removes the target assignment of a member.
+ *
+ * @param memberId The member id.
+ */
+ public void removeTargetAssignment(String memberId) {
+ targetAssignment.remove(memberId);
+ }
+
+ /**
+ * @return An immutable Map containing all the target assignment keyed by member id.
+ */
+ public Map<String, Assignment> targetAssignment() {
+ return Collections.unmodifiableMap(targetAssignment);
+ }
+
+ /**
+ * Returns the current epoch of a partition or -1 if the partition
+ * does not have one.
+ *
+ * @param topicId The topic id.
+ * @param partitionId The partition id.
+ *
+ * @return The epoch or -1.
+ */
+ public int currentPartitionEpoch(
+ Uuid topicId, int partitionId
+ ) {
+ Map<Integer, Integer> partitions = currentPartitionEpoch.get(topicId);
+ if (partitions == null) {
+ return -1;
+ } else {
+ return partitions.getOrDefault(partitionId, -1);
+ }
+ }
+
+ /**
+ * @return An immutable Map of subscription metadata for
+ * each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> subscriptionMetadata() {
+ return Collections.unmodifiableMap(subscribedTopicMetadata);
+ }
+
+ /**
+ * Updates the subscription metadata. This replaces the previous one.
+ *
+ * @param subscriptionMetadata The new subscription metadata.
+ */
+ public void setSubscriptionMetadata(
+ Map<String, TopicMetadata> subscriptionMetadata
+ ) {
+ this.subscribedTopicMetadata.clear();
+ this.subscribedTopicMetadata.putAll(subscriptionMetadata);
+ }
+
+ /**
+ * Computes the subscription metadata based on the current subscription info.
+ *
+ * @param subscribedTopicNames Map of topic names to the number of subscribers.
+ * @param topicsImage The current metadata for all available topics.
+ * @param clusterImage The current metadata for the Kafka cluster.
+ *
+ * @return An immutable map of subscription metadata for each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> computeSubscriptionMetadata(
+ Map<String, Integer> subscribedTopicNames,
+ TopicsImage topicsImage,
+ ClusterImage clusterImage
+ ) {
+ // Create the topic metadata for each subscribed topic.
+ Map<String, TopicMetadata> newSubscriptionMetadata = new HashMap<>(subscribedTopicNames.size());
+
+ subscribedTopicNames.forEach((topicName, count) -> {
+ TopicImage topicImage = topicsImage.getTopic(topicName);
+ if (topicImage != null) {
+ Map<Integer, Set<String>> partitionRacks = new HashMap<>();
+ topicImage.partitions().forEach((partition, partitionRegistration) -> {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ Optional<String> rackOptional = clusterImage.broker(replica).rack();
+ // Only add the rack if it is available for the broker/replica.
+ rackOptional.ifPresent(racks::add);
+ }
+ // If rack information is unavailable for all replicas of this partition,
+ // no corresponding entry will be stored for it in the map.
+ if (!racks.isEmpty())
+ partitionRacks.put(partition, racks);
+ });
+
+ newSubscriptionMetadata.put(topicName, new TopicMetadata(
+ topicImage.id(),
+ topicImage.name(),
+ topicImage.partitions().size(),
+ partitionRacks)
+ );
+ }
+ });
+
+ return Collections.unmodifiableMap(newSubscriptionMetadata);
+ }
+
+ /**
+ * Updates the metadata refresh deadline.
+ *
+ * @param deadlineMs The deadline in milliseconds.
+ * @param groupEpoch The associated group epoch.
+ */
+ public void setMetadataRefreshDeadline(
+ long deadlineMs,
+ int groupEpoch
+ ) {
+ this.metadataRefreshDeadline = new DeadlineAndEpoch(deadlineMs, groupEpoch);
+ }
+
+ /**
+ * Requests a metadata refresh.
+ */
+ public void requestMetadataRefresh() {
+ this.metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+ }
+
+ /**
+ * Checks if a metadata refresh is required. A refresh is required in two cases:
+ * 1) The deadline is smaller or equal to the current time;
+ * 2) The group epoch associated with the deadline is larger than
+ * the current group epoch. This means that the operations which updated
+ * the deadline failed.
+ *
+ * @param currentTimeMs The current time in milliseconds.
+ * @return A boolean indicating whether a refresh is required or not.
+ */
+ public boolean hasMetadataExpired(long currentTimeMs) {
+ return currentTimeMs >= metadataRefreshDeadline.deadlineMs || groupEpoch() < metadataRefreshDeadline.epoch;
+ }
+
+ /**
+ * @return The metadata refresh deadline.
+ */
+ public DeadlineAndEpoch metadataRefreshDeadline() {
+ return metadataRefreshDeadline;
+ }
+
+ /**
+ * Validates the OffsetCommit request.
+ *
+ * @param memberId The member id.
+ * @param groupInstanceId The group instance id.
+ * @param memberEpoch The member epoch.
+ * @param isTransactional Whether the offset commit is transactional or not. It has no
+ * impact when a consumer group is used.
+ */
+ @Override
+ public void validateOffsetCommit(
+ String memberId,
+ String groupInstanceId,
+ int memberEpoch,
+ boolean isTransactional
+ ) throws UnknownMemberIdException, StaleMemberEpochException {
+ // When the member epoch is -1, the request comes from either the admin client
+ // or a consumer which does not use the group management facility. In this case,
+ // the request can commit offsets if the group is empty.
+ if (memberEpoch < 0 && members().isEmpty()) return;
+
+ final GroupMember member = getOrMaybeCreateMember(memberId, false);
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ }
+
+ /**
+ * Validates the OffsetFetch request.
+ *
+ * @param memberId The member id for consumer groups.
+ * @param memberEpoch The member epoch for consumer groups.
+ * @param lastCommittedOffset The last committed offsets in the timeline.
+ */
+ @Override
+ public void validateOffsetFetch(
+ String memberId,
+ int memberEpoch,
+ long lastCommittedOffset
+ ) throws UnknownMemberIdException, StaleMemberEpochException {
+ // When the member id is null and the member epoch is -1, the request either comes
+ // from the admin client or from a client which does not provide them. In this case,
+ // the fetch request is accepted.
+ if (memberId == null && memberEpoch < 0) return;
+
+ final GroupMember member = members.get(memberId, lastCommittedOffset);
+ if (member == null) {
+ throw new UnknownMemberIdException(String.format("Member %s is not a member of group %s.",
+ memberId, groupId));
+ }
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ }
+
+ /**
+ * Validates the OffsetDelete request.
+ */
+ @Override
+ public void validateOffsetDelete() {
+ // Do nothing.
+ }
+
+ /**
+ * See {@link org.apache.kafka.coordinator.group.OffsetExpirationCondition}
+ *
+ * @return The offset expiration condition for the group or Empty if no such condition exists.
+ */
+ @Override
+ public Optional<OffsetExpirationCondition> offsetExpirationCondition() {
+ return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata -> offsetAndMetadata.commitTimestampMs));
+ }
+
+ /**
+ * Throws a StaleMemberEpochException if the received member epoch does not match
+ * the expected member epoch.
+ */
+ private void validateMemberEpoch(
+ int receivedMemberEpoch,
+ int expectedMemberEpoch
+ ) throws StaleMemberEpochException {
+ if (receivedMemberEpoch != expectedMemberEpoch) {
+ throw new StaleMemberEpochException(String.format("The received member epoch %d does not match "
+ + "the expected member epoch %d.", receivedMemberEpoch, expectedMemberEpoch));
+ }
+ }
+
+ /**
+ * Updates the subscribed topic names count.
+ * The subscription type is updated as a consequence.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ protected void maybeUpdateSubscribedTopicNamesAndGroupSubscriptionType(
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ maybeUpdateSubscribedTopicNames(subscribedTopicNames, oldMember, newMember);
+ subscriptionType.set(subscriptionType(subscribedTopicNames, members.size()));
+ }
+
+ /**
+ * Updates the subscription count.
+ *
+ * @param subscribedTopicCount The map to update.
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ private static void maybeUpdateSubscribedTopicNames(
+ Map<String, Integer> subscribedTopicCount,
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ if (oldMember != null) {
+ oldMember.subscribedTopicNames().forEach(topicName ->
+ subscribedTopicCount.compute(topicName, ConsumerGroup::decValue)
+ );
+ }
+
+ if (newMember != null) {
+ newMember.subscribedTopicNames().forEach(topicName ->
+ subscribedTopicCount.compute(topicName, ConsumerGroup::incValue)
+ );
+ }
+ }
+
+ /**
+ * Updates the subscription count.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ *
+ * @return Copy of the map of topics to the count of number of subscribers.
+ */
+ public Map<String, Integer> computeSubscribedTopicNames(
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ Map<String, Integer> subscribedTopicNames = new HashMap<>(this.subscribedTopicNames);
+ maybeUpdateSubscribedTopicNames(
+ subscribedTopicNames,
+ oldMember,
+ newMember
+ );
+ return subscribedTopicNames;
+ }
+
+ /**
+ * Compute the subscription type of the consumer group.
+ *
+ * @param subscribedTopicNames A map of topic names to the count of members subscribed to each topic.
+ *
+ * @return {@link SubscriptionType#HOMOGENEOUS} if all members are subscribed to exactly the same topics;
+ * otherwise, {@link SubscriptionType#HETEROGENEOUS}.
+ */
+ public static SubscriptionType subscriptionType(
+ Map<String, Integer> subscribedTopicNames,
+ int numberOfMembers
+ ) {
+ if (subscribedTopicNames.isEmpty()) {
+ return HOMOGENEOUS;
+ }
+
+ for (int subscriberCount : subscribedTopicNames.values()) {
+ if (subscriberCount != numberOfMembers) {
+ return HETEROGENEOUS;
+ }
+ }
+ return HOMOGENEOUS;
+ }
+
+ /**
+ * Decrements value by 1; returns null when reaching zero. This helper is
+ * meant to be used with Map#compute.
+ */
+ protected static Integer decValue(String key, Integer value) {
+ if (value == null) return null;
+ return value == 1 ? null : value - 1;
+ }
+
+ /**
+ * Increments value by 1; This helper is meant to be used with Map#compute.
+ */
+ protected static Integer incValue(String key, Integer value) {
+ return value == null ? 1 : value + 1;
+ }
Review Comment:
We should probably move those two to an utils library.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/TargetAssignmentBuilderTest.java:
##########
@@ -128,10 +129,10 @@ public void updateMemberSubscription(
Optional<String> instanceId,
Optional<String> rackId
) {
- ConsumerGroupMember existingMember = members.get(memberId);
+ GroupMember existingMember = members.get(memberId);
ConsumerGroupMember.Builder builder;
if (existingMember != null) {
- builder = new ConsumerGroupMember.Builder(existingMember);
+ builder = new ConsumerGroupMember.Builder((ConsumerGroupMember) existingMember);
Review Comment:
This is exactly what I don't want to see.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java:
##########
@@ -166,4 +166,20 @@ void validateOffsetFetch(
* @return true if the state includes, false otherwise.
Review Comment:
I would move this one to common too.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMember.java:
##########
@@ -0,0 +1,217 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.coordinator.group.common.MemberState;
+import org.apache.kafka.coordinator.group.generated.ConsumerGroupCurrentMemberAssignmentValue;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Abstract member common for group members.
+ */
+public abstract class GroupMember {
Review Comment:
I would move it to common too. I would also try to find a better name as it is not used by all the group types.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/AbstractGroup.java:
##########
@@ -0,0 +1,625 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.coordinator.group;
+
+import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.StaleMemberEpochException;
+import org.apache.kafka.common.errors.UnknownMemberIdException;
+import org.apache.kafka.common.message.ListGroupsResponseData;
+import org.apache.kafka.coordinator.group.assignor.SubscriptionType;
+import org.apache.kafka.coordinator.group.consumer.Assignment;
+import org.apache.kafka.coordinator.group.consumer.ConsumerGroup;
+import org.apache.kafka.coordinator.group.consumer.TopicMetadata;
+import org.apache.kafka.image.ClusterImage;
+import org.apache.kafka.image.TopicImage;
+import org.apache.kafka.image.TopicsImage;
+import org.apache.kafka.timeline.SnapshotRegistry;
+import org.apache.kafka.timeline.TimelineHashMap;
+import org.apache.kafka.timeline.TimelineInteger;
+import org.apache.kafka.timeline.TimelineObject;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.Set;
+
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HETEROGENEOUS;
+import static org.apache.kafka.coordinator.group.assignor.SubscriptionType.HOMOGENEOUS;
+
+public abstract class AbstractGroup implements Group {
+
+ public static class DeadlineAndEpoch {
+ static final DeadlineAndEpoch EMPTY = new DeadlineAndEpoch(0L, 0);
+
+ public final long deadlineMs;
+ public final int epoch;
+
+ DeadlineAndEpoch(long deadlineMs, int epoch) {
+ this.deadlineMs = deadlineMs;
+ this.epoch = epoch;
+ }
+ }
+
+ /**
+ * The snapshot registry.
+ */
+ protected final SnapshotRegistry snapshotRegistry;
+
+ /**
+ * The group id.
+ */
+ protected final String groupId;
+
+ /**
+ * The group epoch. The epoch is incremented whenever the subscriptions
+ * are updated and it will trigger the computation of a new assignment
+ * for the group.
+ */
+ protected final TimelineInteger groupEpoch;
+
+ /**
+ * The group members.
+ */
+ protected final TimelineHashMap<String, GroupMember> members;
+
+ /**
+ * The number of subscribers per topic.
+ */
+ protected final TimelineHashMap<String, Integer> subscribedTopicNames;
+
+ /**
+ * The metadata associated with each subscribed topic name.
+ */
+ protected final TimelineHashMap<String, TopicMetadata> subscribedTopicMetadata;
+
+ /**
+ * The group's subscription type.
+ * This value is set to Homogeneous by default.
+ */
+ protected final TimelineObject<SubscriptionType> subscriptionType;
+
+ /**
+ * The target assignment epoch. An assignment epoch smaller than the group epoch
+ * means that a new assignment is required. The assignment epoch is updated when
+ * a new assignment is installed.
+ */
+ protected final TimelineInteger targetAssignmentEpoch;
+
+ /**
+ * The target assignment per member id.
+ */
+ protected final TimelineHashMap<String, Assignment> targetAssignment;
+
+ /**
+ * The current partition epoch maps each topic-partitions to their current epoch where
+ * the epoch is the epoch of their owners. When a member revokes a partition, it removes
+ * its epochs from this map. When a member gets a partition, it adds its epochs to this map.
+ */
+ protected final TimelineHashMap<Uuid, TimelineHashMap<Integer, Integer>> currentPartitionEpoch;
+
+ /**
+ * The metadata refresh deadline. It consists of a timestamp in milliseconds together with
+ * the group epoch at the time of setting it. The metadata refresh time is considered as a
+ * soft state (read that it is not stored in a timeline data structure). It is like this
+ * because it is not persisted to the log. The group epoch is here to ensure that the
+ * metadata refresh deadline is invalidated if the group epoch does not correspond to
+ * the current group epoch. This can happen if the metadata refresh deadline is updated
+ * after having refreshed the metadata but the write operation failed. In this case, the
+ * time is not automatically rolled back.
+ */
+ protected DeadlineAndEpoch metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+
+ protected AbstractGroup(
+ SnapshotRegistry snapshotRegistry,
+ String groupId
+ ) {
+ this.snapshotRegistry = Objects.requireNonNull(snapshotRegistry);
+ this.groupId = Objects.requireNonNull(groupId);
+ this.groupEpoch = new TimelineInteger(snapshotRegistry);
+ this.members = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicNames = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscribedTopicMetadata = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.subscriptionType = new TimelineObject<>(snapshotRegistry, HOMOGENEOUS);
+ this.targetAssignmentEpoch = new TimelineInteger(snapshotRegistry);
+ this.targetAssignment = new TimelineHashMap<>(snapshotRegistry, 0);
+ this.currentPartitionEpoch = new TimelineHashMap<>(snapshotRegistry, 0);
+ }
+
+ /**
+ * @return the group formatted as a list group response based on the committed offset.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup(long committedOffset) {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(stateAsString(committedOffset))
+ .setGroupType(type().toString());
+ }
+
+ /**
+ * @return The group id.
+ */
+ @Override
+ public String groupId() {
+ return groupId;
+ }
+
+ /**
+ * @return The group epoch.
+ */
+ public int groupEpoch() {
+ return groupEpoch.get();
+ }
+
+ /**
+ * Sets the group epoch.
+ *
+ * @param groupEpoch The new group epoch.
+ */
+ public void setGroupEpoch(int groupEpoch) {
+ this.groupEpoch.set(groupEpoch);
+ maybeUpdateGroupState();
+ }
+
+ /**
+ * @return The target assignment epoch.
+ */
+ public int assignmentEpoch() {
+ return targetAssignmentEpoch.get();
+ }
+
+ /**
+ * Sets the assignment epoch.
+ *
+ * @param targetAssignmentEpoch The new assignment epoch.
+ */
+ public void setTargetAssignmentEpoch(int targetAssignmentEpoch) {
+ this.targetAssignmentEpoch.set(targetAssignmentEpoch);
+ maybeUpdateGroupState();
+ }
+
+ /**
+ * @return A boolean indicating whether the member exists or not.
+ */
+ @Override
+ public boolean hasMember(String memberId) {
+ return members.containsKey(memberId);
+ }
+
+ /**
+ * @return The number of members.
+ */
+ @Override
+ public int numMembers() {
+ return members.size();
+ }
+
+ /**
+ * @return An immutable Map containing all the members keyed by their id.
+ */
+ public Map<String, GroupMember> members() {
+ return Collections.unmodifiableMap(members);
+ }
+
+ /**
+ * @return An immutable map containing all the subscribed topic names
+ * with the subscribers counts per topic.
+ */
+ public Map<String, Integer> subscribedTopicNames() {
+ return Collections.unmodifiableMap(subscribedTopicNames);
+ }
+
+ /**
+ * Returns true if the group is actively subscribed to the topic.
+ *
+ * @param topic The topic name.
+ *
+ * @return Whether the group is subscribed to the topic.
+ */
+ @Override
+ public boolean isSubscribedToTopic(String topic) {
+ return subscribedTopicNames.containsKey(topic);
+ }
+
+ /**
+ * @return The group's subscription type.
+ */
+ public SubscriptionType subscriptionType() {
+ return subscriptionType.get();
+ }
+
+ /**
+ * Returns the target assignment of the member.
+ *
+ * @return The ConsumerGroupMemberAssignment or an EMPTY one if it does not
+ * exist.
+ */
+ public Assignment targetAssignment(String memberId) {
+ return targetAssignment.getOrDefault(memberId, Assignment.EMPTY);
+ }
+
+ /**
+ * Updates target assignment of a member.
+ *
+ * @param memberId The member id.
+ * @param newTargetAssignment The new target assignment.
+ */
+ public void updateTargetAssignment(String memberId, Assignment newTargetAssignment) {
+ targetAssignment.put(memberId, newTargetAssignment);
+ }
+
+ /**
+ * Removes the target assignment of a member.
+ *
+ * @param memberId The member id.
+ */
+ public void removeTargetAssignment(String memberId) {
+ targetAssignment.remove(memberId);
+ }
+
+ /**
+ * @return An immutable Map containing all the target assignment keyed by member id.
+ */
+ public Map<String, Assignment> targetAssignment() {
+ return Collections.unmodifiableMap(targetAssignment);
+ }
+
+ /**
+ * Returns the current epoch of a partition or -1 if the partition
+ * does not have one.
+ *
+ * @param topicId The topic id.
+ * @param partitionId The partition id.
+ *
+ * @return The epoch or -1.
+ */
+ public int currentPartitionEpoch(
+ Uuid topicId, int partitionId
+ ) {
+ Map<Integer, Integer> partitions = currentPartitionEpoch.get(topicId);
+ if (partitions == null) {
+ return -1;
+ } else {
+ return partitions.getOrDefault(partitionId, -1);
+ }
+ }
+
+ /**
+ * @return An immutable Map of subscription metadata for
+ * each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> subscriptionMetadata() {
+ return Collections.unmodifiableMap(subscribedTopicMetadata);
+ }
+
+ /**
+ * Updates the subscription metadata. This replaces the previous one.
+ *
+ * @param subscriptionMetadata The new subscription metadata.
+ */
+ public void setSubscriptionMetadata(
+ Map<String, TopicMetadata> subscriptionMetadata
+ ) {
+ this.subscribedTopicMetadata.clear();
+ this.subscribedTopicMetadata.putAll(subscriptionMetadata);
+ }
+
+ /**
+ * Computes the subscription metadata based on the current subscription info.
+ *
+ * @param subscribedTopicNames Map of topic names to the number of subscribers.
+ * @param topicsImage The current metadata for all available topics.
+ * @param clusterImage The current metadata for the Kafka cluster.
+ *
+ * @return An immutable map of subscription metadata for each topic that the consumer group is subscribed to.
+ */
+ public Map<String, TopicMetadata> computeSubscriptionMetadata(
+ Map<String, Integer> subscribedTopicNames,
+ TopicsImage topicsImage,
+ ClusterImage clusterImage
+ ) {
+ // Create the topic metadata for each subscribed topic.
+ Map<String, TopicMetadata> newSubscriptionMetadata = new HashMap<>(subscribedTopicNames.size());
+
+ subscribedTopicNames.forEach((topicName, count) -> {
+ TopicImage topicImage = topicsImage.getTopic(topicName);
+ if (topicImage != null) {
+ Map<Integer, Set<String>> partitionRacks = new HashMap<>();
+ topicImage.partitions().forEach((partition, partitionRegistration) -> {
+ Set<String> racks = new HashSet<>();
+ for (int replica : partitionRegistration.replicas) {
+ Optional<String> rackOptional = clusterImage.broker(replica).rack();
+ // Only add the rack if it is available for the broker/replica.
+ rackOptional.ifPresent(racks::add);
+ }
+ // If rack information is unavailable for all replicas of this partition,
+ // no corresponding entry will be stored for it in the map.
+ if (!racks.isEmpty())
+ partitionRacks.put(partition, racks);
+ });
+
+ newSubscriptionMetadata.put(topicName, new TopicMetadata(
+ topicImage.id(),
+ topicImage.name(),
+ topicImage.partitions().size(),
+ partitionRacks)
+ );
+ }
+ });
+
+ return Collections.unmodifiableMap(newSubscriptionMetadata);
+ }
+
+ /**
+ * Updates the metadata refresh deadline.
+ *
+ * @param deadlineMs The deadline in milliseconds.
+ * @param groupEpoch The associated group epoch.
+ */
+ public void setMetadataRefreshDeadline(
+ long deadlineMs,
+ int groupEpoch
+ ) {
+ this.metadataRefreshDeadline = new DeadlineAndEpoch(deadlineMs, groupEpoch);
+ }
+
+ /**
+ * Requests a metadata refresh.
+ */
+ public void requestMetadataRefresh() {
+ this.metadataRefreshDeadline = DeadlineAndEpoch.EMPTY;
+ }
+
+ /**
+ * Checks if a metadata refresh is required. A refresh is required in two cases:
+ * 1) The deadline is smaller or equal to the current time;
+ * 2) The group epoch associated with the deadline is larger than
+ * the current group epoch. This means that the operations which updated
+ * the deadline failed.
+ *
+ * @param currentTimeMs The current time in milliseconds.
+ * @return A boolean indicating whether a refresh is required or not.
+ */
+ public boolean hasMetadataExpired(long currentTimeMs) {
+ return currentTimeMs >= metadataRefreshDeadline.deadlineMs || groupEpoch() < metadataRefreshDeadline.epoch;
+ }
+
+ /**
+ * @return The metadata refresh deadline.
+ */
+ public DeadlineAndEpoch metadataRefreshDeadline() {
+ return metadataRefreshDeadline;
+ }
+
+ /**
+ * Validates the OffsetCommit request.
+ *
+ * @param memberId The member id.
+ * @param groupInstanceId The group instance id.
+ * @param memberEpoch The member epoch.
+ * @param isTransactional Whether the offset commit is transactional or not. It has no
+ * impact when a consumer group is used.
+ */
+ @Override
+ public void validateOffsetCommit(
+ String memberId,
+ String groupInstanceId,
+ int memberEpoch,
+ boolean isTransactional
+ ) throws UnknownMemberIdException, StaleMemberEpochException {
+ // When the member epoch is -1, the request comes from either the admin client
+ // or a consumer which does not use the group management facility. In this case,
+ // the request can commit offsets if the group is empty.
+ if (memberEpoch < 0 && members().isEmpty()) return;
+
+ final GroupMember member = getOrMaybeCreateMember(memberId, false);
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ }
+
+ /**
+ * Validates the OffsetFetch request.
+ *
+ * @param memberId The member id for consumer groups.
+ * @param memberEpoch The member epoch for consumer groups.
+ * @param lastCommittedOffset The last committed offsets in the timeline.
+ */
+ @Override
+ public void validateOffsetFetch(
+ String memberId,
+ int memberEpoch,
+ long lastCommittedOffset
+ ) throws UnknownMemberIdException, StaleMemberEpochException {
+ // When the member id is null and the member epoch is -1, the request either comes
+ // from the admin client or from a client which does not provide them. In this case,
+ // the fetch request is accepted.
+ if (memberId == null && memberEpoch < 0) return;
+
+ final GroupMember member = members.get(memberId, lastCommittedOffset);
+ if (member == null) {
+ throw new UnknownMemberIdException(String.format("Member %s is not a member of group %s.",
+ memberId, groupId));
+ }
+ validateMemberEpoch(memberEpoch, member.memberEpoch());
+ }
+
+ /**
+ * Validates the OffsetDelete request.
+ */
+ @Override
+ public void validateOffsetDelete() {
+ // Do nothing.
+ }
+
+ /**
+ * See {@link org.apache.kafka.coordinator.group.OffsetExpirationCondition}
+ *
+ * @return The offset expiration condition for the group or Empty if no such condition exists.
+ */
+ @Override
+ public Optional<OffsetExpirationCondition> offsetExpirationCondition() {
+ return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata -> offsetAndMetadata.commitTimestampMs));
+ }
+
+ /**
+ * Throws a StaleMemberEpochException if the received member epoch does not match
+ * the expected member epoch.
+ */
+ private void validateMemberEpoch(
+ int receivedMemberEpoch,
+ int expectedMemberEpoch
+ ) throws StaleMemberEpochException {
+ if (receivedMemberEpoch != expectedMemberEpoch) {
+ throw new StaleMemberEpochException(String.format("The received member epoch %d does not match "
+ + "the expected member epoch %d.", receivedMemberEpoch, expectedMemberEpoch));
+ }
+ }
+
+ /**
+ * Updates the subscribed topic names count.
+ * The subscription type is updated as a consequence.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ protected void maybeUpdateSubscribedTopicNamesAndGroupSubscriptionType(
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ maybeUpdateSubscribedTopicNames(subscribedTopicNames, oldMember, newMember);
+ subscriptionType.set(subscriptionType(subscribedTopicNames, members.size()));
+ }
+
+ /**
+ * Updates the subscription count.
+ *
+ * @param subscribedTopicCount The map to update.
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ */
+ private static void maybeUpdateSubscribedTopicNames(
+ Map<String, Integer> subscribedTopicCount,
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ if (oldMember != null) {
+ oldMember.subscribedTopicNames().forEach(topicName ->
+ subscribedTopicCount.compute(topicName, ConsumerGroup::decValue)
+ );
+ }
+
+ if (newMember != null) {
+ newMember.subscribedTopicNames().forEach(topicName ->
+ subscribedTopicCount.compute(topicName, ConsumerGroup::incValue)
+ );
+ }
+ }
+
+ /**
+ * Updates the subscription count.
+ *
+ * @param oldMember The old member.
+ * @param newMember The new member.
+ *
+ * @return Copy of the map of topics to the count of number of subscribers.
+ */
+ public Map<String, Integer> computeSubscribedTopicNames(
+ GroupMember oldMember,
+ GroupMember newMember
+ ) {
+ Map<String, Integer> subscribedTopicNames = new HashMap<>(this.subscribedTopicNames);
+ maybeUpdateSubscribedTopicNames(
+ subscribedTopicNames,
+ oldMember,
+ newMember
+ );
+ return subscribedTopicNames;
+ }
+
+ /**
+ * Compute the subscription type of the consumer group.
+ *
+ * @param subscribedTopicNames A map of topic names to the count of members subscribed to each topic.
+ *
+ * @return {@link SubscriptionType#HOMOGENEOUS} if all members are subscribed to exactly the same topics;
+ * otherwise, {@link SubscriptionType#HETEROGENEOUS}.
+ */
+ public static SubscriptionType subscriptionType(
+ Map<String, Integer> subscribedTopicNames,
+ int numberOfMembers
+ ) {
+ if (subscribedTopicNames.isEmpty()) {
+ return HOMOGENEOUS;
+ }
+
+ for (int subscriberCount : subscribedTopicNames.values()) {
+ if (subscriberCount != numberOfMembers) {
+ return HETEROGENEOUS;
+ }
+ }
+ return HOMOGENEOUS;
+ }
+
+ /**
+ * Decrements value by 1; returns null when reaching zero. This helper is
+ * meant to be used with Map#compute.
+ */
+ protected static Integer decValue(String key, Integer value) {
+ if (value == null) return null;
+ return value == 1 ? null : value - 1;
+ }
+
+ /**
+ * Increments value by 1; This helper is meant to be used with Map#compute.
+ */
+ protected static Integer incValue(String key, Integer value) {
+ return value == null ? 1 : value + 1;
+ }
+
+ /**
+ * Gets or creates a member.
+ *
+ * @param memberId The member id.
+ * @param createIfNotExists Booleans indicating whether the member must be
+ * created if it does not exist.
+ *
+ * @return A ConsumerGroupMember.
+ */
+ public abstract GroupMember getOrMaybeCreateMember(String memberId, boolean createIfNotExists);
Review Comment:
I don't like that we return `GroupMember` here. When I have a ConsumerGroup, I want it to return and take ConsumerGroupMember because I don't want to cast all the time. Would it be possible to parameterize the base class with something like `<T extends GroupMember>` and then use T?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-16822: Abstract consumer group to share functionality with share group (KIP-932) [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #16054:
URL: https://github.com/apache/kafka/pull/16054
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org