hachikuji commented on a change in pull request #8238:
URL: https://github.com/apache/kafka/pull/8238#discussion_r431596763
##########
File path: clients/src/main/resources/common/message/ListGroupsResponse.json
##########
@@ -34,7 +36,9 @@
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType":
"groupId",
"about": "The group ID." },
{ "name": "ProtocolType", "type": "string", "versions": "0+",
- "about": "The group protocol type." }
+ "about": "The group protocol type." },
+ { "name": "GroupState", "type": "string", "versions": "4+",
"nullableVersions": "0+", "ignorable": true, "default": "null",
Review comment:
Is it intentional to use nullable versions 0+? I'm surprised the
generator doesn't fail.
##########
File path: clients/src/test/java/org/apache/kafka/clients/MockClient.java
##########
@@ -215,15 +215,22 @@ public void send(ClientRequest request, long now) {
AbstractRequest.Builder<?> builder = request.requestBuilder();
short version =
nodeApiVersions.latestUsableVersion(request.apiKey(),
builder.oldestAllowedVersion(),
builder.latestAllowedVersion());
- AbstractRequest abstractRequest =
request.requestBuilder().build(version);
- if (!futureResp.requestMatcher.matches(abstractRequest))
- throw new IllegalStateException("Request matcher did not match
next-in-line request " + abstractRequest + " with prepared response " +
futureResp.responseBody);
UnsupportedVersionException unsupportedVersionException = null;
if (futureResp.isUnsupportedRequest)
- unsupportedVersionException = new
UnsupportedVersionException("Api " +
- request.apiKey() + " with version " + version);
-
+ unsupportedVersionException = new UnsupportedVersionException(
+ "Api " + request.apiKey() + " with version " +
version);
+ try {
+ AbstractRequest abstractRequest =
request.requestBuilder().build(version);
Review comment:
It's a little odd that we go on and match the request if it is expected
to raise an unsupported version error. Do we have test cases that depend on
this? In `NetworkClient`, if we hit an unsupported version error, the response
body would be null.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupListing.java
##########
@@ -17,22 +17,30 @@
package org.apache.kafka.clients.admin;
+import java.util.Objects;
+import java.util.Optional;
+
+import org.apache.kafka.common.ConsumerGroupState;
+
/**
* A listing of a consumer group in the cluster.
*/
public class ConsumerGroupListing {
private final String groupId;
private final boolean isSimpleConsumerGroup;
+ private final Optional<ConsumerGroupState> state;
/**
* 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
*/
- public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup)
{
+ public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup,
Optional<ConsumerGroupState> state) {
Review comment:
Hmm, seems this is a public class. Would it be safer to add a
constructor overload?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/ListConsumerGroupsOptions.java
##########
@@ -26,4 +31,30 @@
*/
@InterfaceStability.Evolving
public class ListConsumerGroupsOptions extends
AbstractOptions<ListConsumerGroupsOptions> {
+
+ private Set<ConsumerGroupState> states = Collections.emptySet();
+
+ /**
+ * Only groups in these states will be returned by listConsumerGroups()
+ * If not set, all groups are returned with their states
+ */
+ public ListConsumerGroupsOptions inStates(Set<ConsumerGroupState> states) {
+ this.states = (states == null) ? Collections.emptySet() : new
HashSet<>(states);
+ return this;
+ }
+
+ /**
+ * All groups with their states will be returned by listConsumerGroups()
+ */
+ public ListConsumerGroupsOptions inAnyState() {
Review comment:
Do we need this API? Seems this is the default behavior.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3052,14 +3052,23 @@ void handleResponse(AbstractResponse abstractResponse) {
runnable.call(new Call("listConsumerGroups", deadline, new
ConstantNodeIdProvider(node.id())) {
@Override
ListGroupsRequest.Builder createRequest(int timeoutMs)
{
- return new ListGroupsRequest.Builder(new
ListGroupsRequestData());
+ List<String> states = (options.states().isEmpty())
Review comment:
nit: I think the parenthesis are unnecessary?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/ConsumerGroupListing.java
##########
@@ -17,22 +17,30 @@
package org.apache.kafka.clients.admin;
+import java.util.Objects;
+import java.util.Optional;
+
+import org.apache.kafka.common.ConsumerGroupState;
+
/**
* A listing of a consumer group in the cluster.
*/
public class ConsumerGroupListing {
private final String groupId;
private final boolean isSimpleConsumerGroup;
+ private final Optional<ConsumerGroupState> state;
/**
* 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
*/
- public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup)
{
+ public ConsumerGroupListing(String groupId, boolean isSimpleConsumerGroup,
Optional<ConsumerGroupState> state) {
this.groupId = groupId;
this.isSimpleConsumerGroup = isSimpleConsumerGroup;
+ this.state = state;
Review comment:
nit: maybe use `requireNonNull`?
##########
File path: core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
##########
@@ -1004,9 +1061,9 @@ object ConsumerGroupCommand extends Logging {
val offsetsOpt = parser.accepts("offsets", OffsetsDoc)
.availableIf(describeOpt)
val stateOpt = parser.accepts("state", StateDoc)
- .availableIf(describeOpt)
-
- parser.mutuallyExclusive(membersOpt, offsetsOpt, stateOpt)
+ .availableIf(describeOpt, listOpt)
+ .withOptionalArg()
Review comment:
Since we have an optional argument, do we need to validate that it is
not provided when `--describe` is used?
##########
File path: core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala
##########
@@ -178,12 +200,44 @@ object ConsumerGroupCommand extends Logging {
} else None
}
- def listGroups(): List[String] = {
+ def listGroups(): Unit = {
+ if (opts.options.has(opts.stateOpt)) {
+ val stateValue = opts.options.valueOf(opts.stateOpt)
+ val states = if (stateValue == null || stateValue.isEmpty)
+ allStates
Review comment:
Seems we need to update this now that we use empty list to request all
states.
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -1312,6 +1324,79 @@ public void testListConsumerGroupsMetadataFailure()
throws Exception {
}
}
+ @Test
+ public void testListConsumerGroupsWithStates() throws Exception {
+ try (AdminClientUnitTestEnv env = new
AdminClientUnitTestEnv(mockCluster(1, 0))) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+
+
env.kafkaClient().prepareResponse(prepareMetadataResponse(env.cluster(),
Errors.NONE));
+
+ env.kafkaClient().prepareResponseFrom(
+ new ListGroupsResponse(new ListGroupsResponseData()
+ .setErrorCode(Errors.NONE.code())
+ .setGroups(Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group-1")
+
.setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState("Stable"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group-2")
+ .setGroupState("Empty")))),
+ env.cluster().nodeById(0));
+
+ final ListConsumerGroupsOptions options = new
ListConsumerGroupsOptions().inAnyState();
+ final ListConsumerGroupsResult result =
env.adminClient().listConsumerGroups(options);
+ Collection<ConsumerGroupListing> listings = result.valid().get();
+
+ assertEquals(2, listings.size());
+ List<ConsumerGroupListing> expected = new ArrayList<>();
+ expected.add(new ConsumerGroupListing("group-2", true,
Optional.of(ConsumerGroupState.EMPTY)));
+ expected.add(new ConsumerGroupListing("group-1", false,
Optional.of(ConsumerGroupState.STABLE)));
+ assertEquals(expected, listings);
+ assertEquals(0, result.errors().get().size());
+ }
+ }
+
+ @Test
+ public void testListConsumerGroupsWithStatesOlderBrokerVersion() throws
Exception {
+ ApiVersion listGroupV3 = new ApiVersion(ApiKeys.LIST_GROUPS.id,
(short) 0, (short) 3);
+ try (AdminClientUnitTestEnv env = new
AdminClientUnitTestEnv(mockCluster(1, 0))) {
+
env.kafkaClient().setNodeApiVersions(NodeApiVersions.create(Collections.singletonList(listGroupV3)));
+
+
env.kafkaClient().prepareResponse(prepareMetadataResponse(env.cluster(),
Errors.NONE));
+
+ // Check we can list groups with older broker if we don't specify
states
+ env.kafkaClient().prepareResponseFrom(
+ new ListGroupsResponse(new ListGroupsResponseData()
+ .setErrorCode(Errors.NONE.code())
+ .setGroups(Collections.singletonList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group-1")
+
.setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)))),
+ env.cluster().nodeById(0));
+ ListConsumerGroupsOptions options = new
ListConsumerGroupsOptions().inAnyState();
+ ListConsumerGroupsResult result =
env.adminClient().listConsumerGroups(options);
+ Collection<ConsumerGroupListing> listing = result.all().get();
+ assertEquals(1, listing.size());
+ List<ConsumerGroupListing> expected =
Collections.singletonList(new ConsumerGroupListing("group-1", false,
Optional.empty()));
+ assertEquals(expected, listing);
+
+ // But we cannot set a state filter with older broker
+
env.kafkaClient().prepareResponse(prepareMetadataResponse(env.cluster(),
Errors.NONE));
+ env.kafkaClient().prepareUnsupportedVersionResponse(
+ body -> body instanceof ListGroupsRequest);
+
+ options = new
ListConsumerGroupsOptions().inStates(Collections.singleton(ConsumerGroupState.STABLE));
+ result = env.adminClient().listConsumerGroups(options);
+ try {
+ result.all().get();
+ fail("Should have thrown");
Review comment:
nit: you can use `TestUtils.assertFutureThrows`
##########
File path: core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala
##########
@@ -797,12 +797,17 @@ class GroupCoordinator(val brokerId: Int,
}
}
- def handleListGroups(): (Errors, List[GroupOverview]) = {
+ def handleListGroups(states: List[String]): (Errors, List[GroupOverview]) = {
Review comment:
Could this be a `Set`?
##########
File path: clients/src/main/resources/common/message/ListGroupsRequest.json
##########
@@ -20,8 +20,14 @@
// Version 1 and 2 are the same as version 0.
//
// Version 3 is the first flexible version.
- "validVersions": "0-3",
+ //
+ // Version 4 adds the StatesFilter field (KIP-518).
+ "validVersions": "0-4",
"flexibleVersions": "3+",
"fields": [
+ { "name": "StatesFilter", "type": "[]string", "versions": "4+",
"nullableVersions": "4+", "default": "null",
+ "about": "The states of the groups we want to list. If empty or null,
all groups are returned with their state.", "fields": [
+ { "name": "Name", "type": "string", "versions": "4+", "about": "The name
of the group state" }
Review comment:
Seems this is ignored because the type of `StatesFilter` is `[]string`.
Sort of feels like a bug in the schema validation that this gets ignored. In
any case, should we remove it?
##########
File path: clients/src/main/resources/common/message/ListGroupsRequest.json
##########
@@ -20,8 +20,14 @@
// Version 1 and 2 are the same as version 0.
//
// Version 3 is the first flexible version.
- "validVersions": "0-3",
+ //
+ // Version 4 adds the StatesFilter field (KIP-518).
+ "validVersions": "0-4",
"flexibleVersions": "3+",
"fields": [
+ { "name": "StatesFilter", "type": "[]string", "versions": "4+",
"nullableVersions": "4+", "default": "null",
Review comment:
The behavior of null is identical to an empty list in terms of
semantics. However, if an empty list is provided, the request cannot be
serialized with older versions because the field is marked as ignorable. Would
it be simpler if we made this field non-nullable and used the default of an
empty array so that we don't have this inconsistency?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
