Skip to content

KAFKA-17949: Introduce GroupState and replace ShareGroupState #17763

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
omkreddy merged 7 commits into from
Nov 19, 2024
Merged

KAFKA-17949: Introduce GroupState and replace ShareGroupState #17763

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
b3ce8b3
KAFKA-17949: Introduce GroupState
AndrewJSchofield Nov 11, 2024
ac87bea
KAFKA-17949: Introduce GroupState and remove ShareGroupState
AndrewJSchofield Nov 12, 2024
6e4209f
Spotless
AndrewJSchofield Nov 12, 2024
0831268
Deprecate ConsumerGroupState and associated methods
AndrewJSchofield Nov 13, 2024
1c51c60
Merge remote-tracking branch 'upstream/trunk' into KAFKA-17949
AndrewJSchofield Nov 15, 2024
4a9714e
Merge remote-tracking branch 'upstream/trunk' into KAFKA-17949
AndrewJSchofield Nov 18, 2024
d5745da
Spotless
AndrewJSchofield Nov 19, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 66 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupDescription.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.kafka.clients.admin;

import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.GroupState;
import org.apache.kafka.common.GroupType;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.acl.AclOperation;
Expand All @@ -38,10 +39,14 @@ public class ConsumerGroupDescription {
private final Collection<MemberDescription> members;
private final String partitionAssignor;
private final GroupType type;
private final ConsumerGroupState state;
private final GroupState groupState;
private final Node coordinator;
private final Set<AclOperation> authorizedOperations;

/**
* @deprecated Since 4.0. Use {@link #ConsumerGroupDescription(String, boolean, Collection, String, GroupState, Node)}.
*/
@Deprecated
public ConsumerGroupDescription(String groupId,
boolean isSimpleConsumerGroup,
Collection<MemberDescription> members,
Expand All @@ -51,6 +56,10 @@ public ConsumerGroupDescription(String groupId,
this(groupId, isSimpleConsumerGroup, members, partitionAssignor, state, coordinator, Collections.emptySet());
}

/**
* @deprecated Since 4.0. Use {@link #ConsumerGroupDescription(String, boolean, Collection, String, GroupState, Node, Set)}.
*/
@Deprecated
public ConsumerGroupDescription(String groupId,
boolean isSimpleConsumerGroup,
Collection<MemberDescription> members,
Expand All @@ -61,6 +70,10 @@ public ConsumerGroupDescription(String groupId,
this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, state, coordinator, authorizedOperations);
}

/**
* @deprecated Since 4.0. Use {@link #ConsumerGroupDescription(String, boolean, Collection, String, GroupType, GroupState, Node, Set)}.
*/
@Deprecated
public ConsumerGroupDescription(String groupId,
boolean isSimpleConsumerGroup,
Collection<MemberDescription> members,
Expand All @@ -75,7 +88,45 @@ public ConsumerGroupDescription(String groupId,
Collections.unmodifiableList(new ArrayList<>(members));
this.partitionAssignor = partitionAssignor == null ? "" : partitionAssignor;
this.type = type;
this.state = state;
this.groupState = GroupState.parse(state.name());
this.coordinator = coordinator;
this.authorizedOperations = authorizedOperations;
}

public ConsumerGroupDescription(String groupId,
boolean isSimpleConsumerGroup,
Collection<MemberDescription> members,
String partitionAssignor,
GroupState groupState,
Node coordinator) {
this(groupId, isSimpleConsumerGroup, members, partitionAssignor, groupState, coordinator, Collections.emptySet());
}

public ConsumerGroupDescription(String groupId,
boolean isSimpleConsumerGroup,
Collection<MemberDescription> members,
String partitionAssignor,
GroupState groupState,
Node coordinator,
Set<AclOperation> authorizedOperations) {
this(groupId, isSimpleConsumerGroup, members, partitionAssignor, GroupType.CLASSIC, groupState, coordinator, authorizedOperations);
}

public ConsumerGroupDescription(String groupId,
boolean isSimpleConsumerGroup,
Collection<MemberDescription> members,
String partitionAssignor,
GroupType type,
GroupState groupState,
Node coordinator,
Set<AclOperation> authorizedOperations) {
this.groupId = groupId == null ? "" : groupId;
this.isSimpleConsumerGroup = isSimpleConsumerGroup;
this.members = members == null ? Collections.emptyList() :
Collections.unmodifiableList(new ArrayList<>(members));
this.partitionAssignor = partitionAssignor == null ? "" : partitionAssignor;
this.type = type;
this.groupState = groupState;
this.coordinator = coordinator;
this.authorizedOperations = authorizedOperations;
}
Expand All @@ -90,14 +141,14 @@ public boolean equals(final Object o) {
Objects.equals(members, that.members) &&
Objects.equals(partitionAssignor, that.partitionAssignor) &&
type == that.type &&
state == that.state &&
groupState == that.groupState &&
Objects.equals(coordinator, that.coordinator) &&
Objects.equals(authorizedOperations, that.authorizedOperations);
}

@Override
public int hashCode() {
return Objects.hash(groupId, isSimpleConsumerGroup, members, partitionAssignor, type, state, coordinator, authorizedOperations);
return Objects.hash(groupId, isSimpleConsumerGroup, members, partitionAssignor, type, groupState, coordinator, authorizedOperations);
}

/**
Expand Down Expand Up @@ -138,9 +189,18 @@ public GroupType type() {

/**
* The consumer group state, or UNKNOWN if the state is too new for us to parse.
* @deprecated Since 4.0. Use {@link #groupState()} instead.
*/
@Deprecated
public ConsumerGroupState state() {
return state;
return ConsumerGroupState.parse(groupState.name());
}

/**
* The group state, or UNKNOWN if the state is too new for us to parse.
*/
public GroupState groupState() {
return groupState;
}

/**
Expand All @@ -164,7 +224,7 @@ public String toString() {
", members=" + members.stream().map(MemberDescription::toString).collect(Collectors.joining(",")) +
", partitionAssignor=" + partitionAssignor +
", type=" + type +
", state=" + state +
", groupState=" + groupState +
", coordinator=" + coordinator +
", authorizedOperations=" + authorizedOperations +
")";
Expand Down
74 changes: 60 additions & 14 deletions clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupListing.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.kafka.clients.admin;

import org.apache.kafka.common.ConsumerGroupState;
import org.apache.kafka.common.GroupState;
import org.apache.kafka.common.GroupType;

import java.util.Objects;
Expand All @@ -29,28 +30,30 @@
public class ConsumerGroupListing {
private final String groupId;
private final boolean isSimpleConsumerGroup;
private final Optional<ConsumerGroupState> state;
private final Optional<GroupState> groupState;
private final Optional<GroupType> type;

/**
* Create an instance with the specified parameters.
*
* @param groupId Group Id
* @param isSimpleConsumerGroup If consumer group is simple or not.
* @param groupId Group Id.
* @param isSimpleConsumerGroup If consumer group is simple or not.
*/
public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup) {
this(groupId, isSimpleConsumerGroup, Optional.empty(), Optional.empty());
this(groupId, Optional.empty(), Optional.empty(), isSimpleConsumerGroup);
}

/**
* Create an instance with the specified parameters.
*
* @param groupId Group Id
* @param isSimpleConsumerGroup If consumer group is simple or not.
* @param state The state of the consumer group
* @param groupId Group Id.
* @param isSimpleConsumerGroup If consumer group is simple or not.
* @param state The state of the consumer group.
* @deprecated Since 4.0. Use {@link #ConsumerGroupListing(String, Optional, boolean)}.
*/
@Deprecated
public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup, Optional<ConsumerGroupState> state) {
this(groupId, isSimpleConsumerGroup, state, Optional.empty());
this(groupId, Objects.requireNonNull(state).map(state0 -> GroupState.parse(state0.name())), Optional.empty(), isSimpleConsumerGroup);
}

/**
Expand All @@ -60,17 +63,51 @@ public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup, Optio
* @param isSimpleConsumerGroup If consumer group is simple or not.
* @param state The state of the consumer group.
* @param type The type of the consumer group.
* @deprecated Since 4.0. Use {@link #ConsumerGroupListing(String, Optional, Optional, boolean)}.
*/
@Deprecated
public ConsumerGroupListing(
String groupId,
boolean isSimpleConsumerGroup,
Optional<ConsumerGroupState> state,
Optional<GroupType> type
) {
this(groupId, Objects.requireNonNull(state).map(state0 -> GroupState.parse(state0.name())), type, isSimpleConsumerGroup);
}

/**
* Create an instance with the specified parameters.
*
* @param groupId Group Id.
* @param groupState The state of the consumer group.
* @param isSimpleConsumerGroup If consumer group is simple or not.
*/
public ConsumerGroupListing(
String groupId,
Optional<GroupState> groupState,
boolean isSimpleConsumerGroup
) {
this(groupId, groupState, Optional.empty(), isSimpleConsumerGroup);
}

/**
* Create an instance with the specified parameters.
*
* @param groupId Group Id.
* @param groupState The state of the consumer group.
* @param type The type of the consumer group.
* @param isSimpleConsumerGroup If consumer group is simple or not.
*/
public ConsumerGroupListing(
String groupId,
Optional<GroupState> groupState,
Optional<GroupType> type,
boolean isSimpleConsumerGroup
) {
this.groupId = groupId;
this.isSimpleConsumerGroup = isSimpleConsumerGroup;
this.state = Objects.requireNonNull(state);
this.groupState = Objects.requireNonNull(groupState);
this.type = Objects.requireNonNull(type);
this.isSimpleConsumerGroup = isSimpleConsumerGroup;
}

/**
Expand All @@ -87,11 +124,20 @@ public boolean isSimpleConsumerGroup() {
return isSimpleConsumerGroup;
}

/**
* Group state
*/
public Optional<GroupState> groupState() {
return groupState;
}

/**
* Consumer Group state
* @deprecated Since 4.0. Use {@link #groupState()}.
*/
@Deprecated
public Optional<ConsumerGroupState> state() {
return state;
return groupState.map(state0 -> ConsumerGroupState.parse(state0.name()));
}

/**
Expand All @@ -108,14 +154,14 @@ public String toString() {
return "(" +
"groupId='" + groupId + '\'' +
", isSimpleConsumerGroup=" + isSimpleConsumerGroup +
", state=" + state +
", groupState=" + groupState +
", type=" + type +
')';
}

@Override
public int hashCode() {
return Objects.hash(groupId, isSimpleConsumerGroup(), state, type);
return Objects.hash(groupId, isSimpleConsumerGroup(), groupState, type);
}

@Override
Expand All @@ -125,7 +171,7 @@ public boolean equals(Object o) {
ConsumerGroupListing that = (ConsumerGroupListing) o;
return isSimpleConsumerGroup() == that.isSimpleConsumerGroup() &&
Objects.equals(groupId, that.groupId) &&
Objects.equals(state, that.state) &&
Objects.equals(groupState, that.groupState) &&
Objects.equals(type, that.type);
}
}
33 changes: 27 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/admin/GroupListing.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,30 +17,36 @@

package org.apache.kafka.clients.admin;

import org.apache.kafka.common.GroupState;
import org.apache.kafka.common.GroupType;
import org.apache.kafka.common.annotation.InterfaceStability;

import java.util.Objects;
import java.util.Optional;

/**
* A listing of a group in the cluster.
*/
@InterfaceStability.Evolving
public class GroupListing {
private final String groupId;
private final Optional<GroupType> type;
private final String protocol;
private final Optional<GroupState> groupState;

/**
* Create an instance with the specified parameters.
*
* @param groupId Group Id
* @param type Group type
* @param protocol Protocol
* @param groupId Group Id
* @param type Group type
* @param protocol Protocol
* @param groupState Group state
*/
public GroupListing(String groupId, Optional<GroupType> type, String protocol) {
public GroupListing(String groupId, Optional<GroupType> type, String protocol, Optional<GroupState> groupState) {
this.groupId = groupId;
this.type = Objects.requireNonNull(type);
this.protocol = protocol;
this.groupState = groupState;
}

/**
Expand Down Expand Up @@ -75,6 +81,19 @@ public String protocol() {
return protocol;
}

/**
* The group state.
* <p>
* If the broker returns a group state which is not recognised, as might
* happen when talking to a broker with a later version, the state will be
* <code>Optional.of(GroupState.UNKNOWN)</code>.
*
* @return An Optional containing the state, if available.
*/
public Optional<GroupState> groupState() {
return groupState;
}

/**
* If the group is a simple consumer group or not.
*/
Expand All @@ -88,12 +107,13 @@ public String toString() {
"groupId='" + groupId + '\'' +
", type=" + type.map(GroupType::toString).orElse("none") +
", protocol='" + protocol + '\'' +
", groupState=" + groupState.map(GroupState::toString).orElse("none") +
')';
}

@Override
public int hashCode() {
return Objects.hash(groupId, type, protocol);
return Objects.hash(groupId, type, protocol, groupState);
}

@Override
Expand All @@ -103,6 +123,7 @@ public boolean equals(Object o) {
GroupListing that = (GroupListing) o;
return Objects.equals(groupId, that.groupId) &&
Objects.equals(type, that.type) &&
Objects.equals(protocol, that.protocol);
Objects.equals(protocol, that.protocol) &&
Objects.equals(groupState, that.groupState);
}
}
Loading