Browse Source
Reviewers: Manikumar Reddy O <manikumar.reddy@gmail.com>, Jason Gustafson <jason@confluent.io>pull/4499/head
Vahid Hashemian
7 years ago
committed by
Jason Gustafson
22 changed files with 941 additions and 47 deletions
@ -0,0 +1,31 @@
@@ -0,0 +1,31 @@
|
||||
/* |
||||
* 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.common.errors; |
||||
|
||||
public class GroupIdNotFoundException extends ApiException { |
||||
private final String groupId; |
||||
|
||||
public GroupIdNotFoundException(String groupId) { |
||||
super("The group id " + groupId + " was not found"); |
||||
this.groupId = groupId; |
||||
} |
||||
|
||||
public String groupId() { |
||||
return groupId; |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,31 @@
@@ -0,0 +1,31 @@
|
||||
/* |
||||
* 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.common.errors; |
||||
|
||||
public class GroupNotEmptyException extends ApiException { |
||||
private final String groupId; |
||||
|
||||
public GroupNotEmptyException(String groupId) { |
||||
super("The group " + groupId + " is not empty"); |
||||
this.groupId = groupId; |
||||
} |
||||
|
||||
public String groupId() { |
||||
return groupId; |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,117 @@
@@ -0,0 +1,117 @@
|
||||
/* |
||||
* 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.common.requests; |
||||
|
||||
import org.apache.kafka.common.protocol.ApiKeys; |
||||
import org.apache.kafka.common.protocol.Errors; |
||||
import org.apache.kafka.common.protocol.types.ArrayOf; |
||||
import org.apache.kafka.common.protocol.types.Field; |
||||
import org.apache.kafka.common.protocol.types.Schema; |
||||
import org.apache.kafka.common.protocol.types.Struct; |
||||
import org.apache.kafka.common.utils.Utils; |
||||
|
||||
import java.nio.ByteBuffer; |
||||
import java.util.HashMap; |
||||
import java.util.HashSet; |
||||
import java.util.Map; |
||||
import java.util.Set; |
||||
|
||||
import static org.apache.kafka.common.protocol.types.Type.STRING; |
||||
|
||||
public class DeleteGroupsRequest extends AbstractRequest { |
||||
private static final String GROUPS_KEY_NAME = "groups"; |
||||
|
||||
/* DeleteGroups api */ |
||||
private static final Schema DELETE_GROUPS_REQUEST_V0 = new Schema( |
||||
new Field(GROUPS_KEY_NAME, new ArrayOf(STRING), "An array of groups to be deleted.")); |
||||
|
||||
public static Schema[] schemaVersions() { |
||||
return new Schema[]{DELETE_GROUPS_REQUEST_V0}; |
||||
} |
||||
|
||||
private final Set<String> groups; |
||||
|
||||
public static class Builder extends AbstractRequest.Builder<DeleteGroupsRequest> { |
||||
private final Set<String> groups; |
||||
|
||||
public Builder(Set<String> groups) { |
||||
super(ApiKeys.DELETE_GROUPS); |
||||
this.groups = groups; |
||||
} |
||||
|
||||
@Override |
||||
public DeleteGroupsRequest build(short version) { |
||||
return new DeleteGroupsRequest(groups, version); |
||||
} |
||||
|
||||
@Override |
||||
public String toString() { |
||||
StringBuilder bld = new StringBuilder(); |
||||
bld.append("(type=DeleteGroupsRequest"). |
||||
append(", groups=(").append(Utils.join(groups, ", ")).append(")"). |
||||
append(")"); |
||||
return bld.toString(); |
||||
} |
||||
} |
||||
|
||||
private DeleteGroupsRequest(Set<String> groups, short version) { |
||||
super(version); |
||||
this.groups = groups; |
||||
} |
||||
|
||||
public DeleteGroupsRequest(Struct struct, short version) { |
||||
super(version); |
||||
Object[] groupsArray = struct.getArray(GROUPS_KEY_NAME); |
||||
Set<String> groups = new HashSet<>(groupsArray.length); |
||||
for (Object group : groupsArray) |
||||
groups.add((String) group); |
||||
|
||||
this.groups = groups; |
||||
} |
||||
|
||||
@Override |
||||
protected Struct toStruct() { |
||||
Struct struct = new Struct(ApiKeys.DELETE_GROUPS.requestSchema(version())); |
||||
struct.set(GROUPS_KEY_NAME, groups.toArray()); |
||||
return struct; |
||||
} |
||||
|
||||
@Override |
||||
public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) { |
||||
Errors error = Errors.forException(e); |
||||
Map<String, Errors> groupErrors = new HashMap<>(groups.size()); |
||||
for (String group : groups) |
||||
groupErrors.put(group, error); |
||||
|
||||
switch (version()) { |
||||
case 0: |
||||
return new DeleteGroupsResponse(throttleTimeMs, groupErrors); |
||||
default: |
||||
throw new IllegalArgumentException(String.format("Version %d is not valid. Valid versions for %s are 0 to %d", |
||||
version(), ApiKeys.DELETE_GROUPS.name, ApiKeys.DELETE_GROUPS.latestVersion())); |
||||
} |
||||
} |
||||
|
||||
public Set<String> groups() { |
||||
return groups; |
||||
} |
||||
|
||||
public static DeleteGroupsRequest parse(ByteBuffer buffer, short version) { |
||||
return new DeleteGroupsRequest(ApiKeys.DELETE_GROUPS.parseRequest(version, buffer), version); |
||||
} |
||||
|
||||
} |
||||
@ -0,0 +1,129 @@
@@ -0,0 +1,129 @@
|
||||
/* |
||||
* 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.common.requests; |
||||
|
||||
import org.apache.kafka.common.protocol.ApiKeys; |
||||
import org.apache.kafka.common.protocol.Errors; |
||||
import org.apache.kafka.common.protocol.types.ArrayOf; |
||||
import org.apache.kafka.common.protocol.types.Field; |
||||
import org.apache.kafka.common.protocol.types.Schema; |
||||
import org.apache.kafka.common.protocol.types.Struct; |
||||
|
||||
import java.nio.ByteBuffer; |
||||
import java.util.ArrayList; |
||||
import java.util.HashMap; |
||||
import java.util.List; |
||||
import java.util.Map; |
||||
|
||||
import static org.apache.kafka.common.protocol.CommonFields.ERROR_CODE; |
||||
import static org.apache.kafka.common.protocol.CommonFields.GROUP_ID; |
||||
import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS; |
||||
|
||||
public class DeleteGroupsResponse extends AbstractResponse { |
||||
private static final String GROUP_ERROR_CODES_KEY_NAME = "group_error_codes"; |
||||
|
||||
private static final Schema GROUP_ERROR_CODE = new Schema( |
||||
GROUP_ID, |
||||
ERROR_CODE); |
||||
|
||||
private static final Schema DELETE_GROUPS_RESPONSE_V0 = new Schema( |
||||
THROTTLE_TIME_MS, |
||||
new Field(GROUP_ERROR_CODES_KEY_NAME, new ArrayOf(GROUP_ERROR_CODE), "An array of per group error codes.")); |
||||
|
||||
public static Schema[] schemaVersions() { |
||||
return new Schema[]{DELETE_GROUPS_RESPONSE_V0}; |
||||
} |
||||
|
||||
|
||||
/** |
||||
* Possible error codes: |
||||
* |
||||
* COORDINATOR_LOAD_IN_PROGRESS (14) |
||||
* COORDINATOR_NOT_AVAILABLE(15) |
||||
* NOT_COORDINATOR (16) |
||||
* INVALID_GROUP_ID(24) |
||||
* GROUP_AUTHORIZATION_FAILED(30) |
||||
* NON_EMPTY_GROUP(68) |
||||
* GROUP_ID_NOT_FOUND(69) |
||||
*/ |
||||
|
||||
private final Map<String, Errors> errors; |
||||
private final int throttleTimeMs; |
||||
|
||||
public DeleteGroupsResponse(Map<String, Errors> errors) { |
||||
this(DEFAULT_THROTTLE_TIME, errors); |
||||
} |
||||
|
||||
public DeleteGroupsResponse(int throttleTimeMs, Map<String, Errors> errors) { |
||||
this.throttleTimeMs = throttleTimeMs; |
||||
this.errors = errors; |
||||
} |
||||
|
||||
public DeleteGroupsResponse(Struct struct) { |
||||
this.throttleTimeMs = struct.getOrElse(THROTTLE_TIME_MS, DEFAULT_THROTTLE_TIME); |
||||
Object[] groupErrorCodesStructs = struct.getArray(GROUP_ERROR_CODES_KEY_NAME); |
||||
Map<String, Errors> errors = new HashMap<>(); |
||||
for (Object groupErrorCodeStructObj : groupErrorCodesStructs) { |
||||
Struct groupErrorCodeStruct = (Struct) groupErrorCodeStructObj; |
||||
String group = groupErrorCodeStruct.get(GROUP_ID); |
||||
Errors error = Errors.forCode(groupErrorCodeStruct.get(ERROR_CODE)); |
||||
errors.put(group, error); |
||||
} |
||||
|
||||
this.errors = errors; |
||||
} |
||||
|
||||
@Override |
||||
protected Struct toStruct(short version) { |
||||
Struct struct = new Struct(ApiKeys.DELETE_GROUPS.responseSchema(version)); |
||||
struct.setIfExists(THROTTLE_TIME_MS, throttleTimeMs); |
||||
List<Struct> groupErrorCodeStructs = new ArrayList<>(errors.size()); |
||||
for (Map.Entry<String, Errors> groupError : errors.entrySet()) { |
||||
Struct groupErrorCodeStruct = struct.instance(GROUP_ERROR_CODES_KEY_NAME); |
||||
groupErrorCodeStruct.set(GROUP_ID, groupError.getKey()); |
||||
groupErrorCodeStruct.set(ERROR_CODE, groupError.getValue().code()); |
||||
groupErrorCodeStructs.add(groupErrorCodeStruct); |
||||
} |
||||
struct.set(GROUP_ERROR_CODES_KEY_NAME, groupErrorCodeStructs.toArray()); |
||||
return struct; |
||||
} |
||||
|
||||
public int throttleTimeMs() { |
||||
return throttleTimeMs; |
||||
} |
||||
|
||||
public Map<String, Errors> errors() { |
||||
return errors; |
||||
} |
||||
|
||||
public boolean hasError(String group) { |
||||
return errors.containsKey(group) && errors.get(group) != Errors.NONE; |
||||
} |
||||
|
||||
public Errors get(String group) { |
||||
return errors.get(group); |
||||
} |
||||
|
||||
@Override |
||||
public Map<Errors, Integer> errorCounts() { |
||||
return errorCounts(errors); |
||||
} |
||||
|
||||
public static DeleteGroupsResponse parse(ByteBuffer buffer, short version) { |
||||
return new DeleteGroupsResponse(ApiKeys.DELETE_GROUPS.responseSchema(version).read(buffer)); |
||||
} |
||||
} |
||||
@ -0,0 +1,251 @@
@@ -0,0 +1,251 @@
|
||||
/** |
||||
* 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 unit.kafka.admin |
||||
|
||||
import kafka.admin.ConsumerGroupCommandTest |
||||
import kafka.utils.TestUtils |
||||
import org.apache.kafka.common.protocol.Errors |
||||
import org.junit.Assert._ |
||||
import org.junit.Test |
||||
|
||||
class DeleteConsumerGroupTest extends ConsumerGroupCommandTest { |
||||
|
||||
@Test(expected = classOf[joptsimple.OptionException]) |
||||
def testDeleteWithTopicOption() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group, "--topic") |
||||
getConsumerGroupService(cgcArgs) |
||||
fail("Expected an error due to presence of mutually exclusive options") |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteCmdNonExistingGroup() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
val missingGroup = "missing.group" |
||||
|
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", missingGroup) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
val output = TestUtils.grabConsoleOutput(service.deleteGroups()) |
||||
assertTrue(s"The expected error (${Errors.GROUP_ID_NOT_FOUND}) was not detected while deleting consumer group", |
||||
output.contains(s"Group '$missingGroup' could not be deleted due to: ${Errors.GROUP_ID_NOT_FOUND.toString}")) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteNonExistingGroup() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
val missingGroup = "missing.group" |
||||
|
||||
// note the group to be deleted is a different (non-existing) group |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", missingGroup) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
val result = service.deleteGroups() |
||||
assertTrue(s"The expected error (${Errors.GROUP_ID_NOT_FOUND}) was not detected while deleting consumer group", |
||||
result.size == 1 && result.keySet.contains(missingGroup) && result.get(missingGroup).contains(Errors.GROUP_ID_NOT_FOUND)) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteCmdInvalidGroupId() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
val invalidGroupId = "" |
||||
|
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", invalidGroupId) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
val output = TestUtils.grabConsoleOutput(service.deleteGroups()) |
||||
assertTrue(s"The expected error (${Errors.INVALID_GROUP_ID}) was not detected while deleting consumer group", |
||||
output.contains(s"Group '$invalidGroupId' could not be deleted due to: ${Errors.INVALID_GROUP_ID.toString}")) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteInvalidGroupId() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
val invalidGroupId = "" |
||||
|
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", invalidGroupId) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
val result = service.deleteGroups() |
||||
assertTrue(s"The expected error (${Errors.INVALID_GROUP_ID}) was not detected while deleting consumer group", |
||||
result.size == 1 && result.keySet.contains(invalidGroupId) && result.get(invalidGroupId).contains(Errors.INVALID_GROUP_ID)) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteCmdNonEmptyGroup() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
|
||||
// run one consumer in the group |
||||
addConsumerGroupExecutor(numConsumers = 1) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.listGroups().contains(group) |
||||
}, "The group did not initialize as expected.") |
||||
|
||||
val output = TestUtils.grabConsoleOutput(service.deleteGroups()) |
||||
assertTrue(s"The expected error (${Errors.NON_EMPTY_GROUP}) was not detected while deleting consumer group", |
||||
output.contains(s"Group '$group' could not be deleted due to: ${Errors.NON_EMPTY_GROUP}")) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteNonEmptyGroup() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
|
||||
// run one consumer in the group |
||||
addConsumerGroupExecutor(numConsumers = 1) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.listGroups().contains(group) |
||||
}, "The group did not initialize as expected.") |
||||
|
||||
val result = service.deleteGroups() |
||||
assertTrue(s"The expected error (${Errors.NON_EMPTY_GROUP}) was not detected while deleting consumer group", |
||||
result.size == 1 && result.keySet.contains(group) && result.get(group).contains(Errors.NON_EMPTY_GROUP)) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteCmdEmptyGroup() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
|
||||
// run one consumer in the group |
||||
val executor = addConsumerGroupExecutor(numConsumers = 1) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.listGroups().contains(group) |
||||
}, "The group did not initialize as expected.") |
||||
|
||||
executor.shutdown() |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.collectGroupState().state == "Empty" |
||||
}, "The group did become empty as expected.") |
||||
|
||||
val output = TestUtils.grabConsoleOutput(service.deleteGroups()) |
||||
assertTrue(s"The consumer group could not be deleted as expected", |
||||
output.contains(s"Deletion of requested consumer groups ('$group') was successful.")) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteEmptyGroup() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
|
||||
// run one consumer in the group |
||||
val executor = addConsumerGroupExecutor(numConsumers = 1) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.listGroups().contains(group) |
||||
}, "The group did not initialize as expected.") |
||||
|
||||
executor.shutdown() |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.collectGroupState().state == "Empty" |
||||
}, "The group did become empty as expected.") |
||||
|
||||
val result = service.deleteGroups() |
||||
assertTrue(s"The consumer group could not be deleted as expected", |
||||
result.size == 1 && result.keySet.contains(group) && result.get(group).contains(Errors.NONE)) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteCmdWithMixOfSuccessAndError() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
val missingGroup = "missing.group" |
||||
|
||||
// run one consumer in the group |
||||
val executor = addConsumerGroupExecutor(numConsumers = 1) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.listGroups().contains(group) |
||||
}, "The group did not initialize as expected.") |
||||
|
||||
executor.shutdown() |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.collectGroupState().state == "Empty" |
||||
}, "The group did become empty as expected.") |
||||
|
||||
val service2 = getConsumerGroupService(cgcArgs ++ Array("--group", missingGroup)) |
||||
val output = TestUtils.grabConsoleOutput(service2.deleteGroups()) |
||||
assertTrue(s"The consumer group deletion did not work as expected", |
||||
output.contains(s"Group '$missingGroup' could not be deleted due to: ${Errors.GROUP_ID_NOT_FOUND}") && |
||||
output.contains(s"These consumer groups were deleted successfully: '$group'")) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteWithMixOfSuccessAndError() { |
||||
TestUtils.createOffsetsTopic(zkClient, servers) |
||||
val missingGroup = "missing.group" |
||||
|
||||
// run one consumer in the group |
||||
val executor = addConsumerGroupExecutor(numConsumers = 1) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.listGroups().contains(group) |
||||
}, "The group did not initialize as expected.") |
||||
|
||||
executor.shutdown() |
||||
|
||||
TestUtils.waitUntilTrue(() => { |
||||
service.collectGroupState().state == "Empty" |
||||
}, "The group did become empty as expected.") |
||||
|
||||
val service2 = getConsumerGroupService(cgcArgs ++ Array("--group", missingGroup)) |
||||
val result = service2.deleteGroups() |
||||
assertTrue(s"The consumer group deletion did not work as expected", |
||||
result.size == 2 && |
||||
result.keySet.contains(group) && result.get(group).contains(Errors.NONE) && |
||||
result.keySet.contains(missingGroup) && result.get(missingGroup).contains(Errors.GROUP_ID_NOT_FOUND)) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteCmdWithShortInitialization() { |
||||
// run one consumer in the group |
||||
val executor = addConsumerGroupExecutor(numConsumers = 1) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
val output = TestUtils.grabConsoleOutput(service.deleteGroups()) |
||||
assertTrue(s"The consumer group deletion did not work as expected", |
||||
output.contains(s"Group '$group' could not be deleted due to: ${Errors.COORDINATOR_NOT_AVAILABLE}")) |
||||
} |
||||
|
||||
@Test |
||||
def testDeleteWithShortInitialization() { |
||||
// run one consumer in the group |
||||
val executor = addConsumerGroupExecutor(numConsumers = 1) |
||||
val cgcArgs = Array("--bootstrap-server", brokerList, "--delete", "--group", group) |
||||
val service = getConsumerGroupService(cgcArgs) |
||||
|
||||
val result = service.deleteGroups() |
||||
assertTrue(s"The consumer group deletion did not work as expected", |
||||
result.size == 1 && |
||||
result.keySet.contains(group) && result.get(group).contains(Errors.COORDINATOR_NOT_AVAILABLE)) |
||||
} |
||||
} |
||||
Loading…
Reference in new issue