mjsax commented on a change in pull request #8589:
URL: https://github.com/apache/kafka/pull/8589#discussion_r430826643
##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -186,9 +192,19 @@ private void validateNoActiveConsumers(final String
groupId,
final List<MemberDescription> members =
new
ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
if (!members.isEmpty()) {
- throw new IllegalStateException("Consumer group '" + groupId + "'
is still active "
- + "and has following members: " + members + ". "
- + "Make sure to stop all running application instances
before running the reset tool.");
+ if (options.has(forceOption)) {
+ System.out.println("Force deleting all active members in the
group: " + groupId);
+ try {
+ adminClient.removeMembersFromConsumerGroup(groupId, new
RemoveMembersFromConsumerGroupOptions()).all().get();
+ } catch (Exception e) {
+ throw e;
+ }
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId +
"' is still active "
+ + "and has following members: " + members + ". "
+ + "Make sure to stop all running application instances
before running the reset tool." +
+ "Try set '--force' in the cmdline to force delete
active members.");
Review comment:
```suggestion
"You can use option '--force' to remove active
members from the group.");
```
##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -186,9 +192,19 @@ private void validateNoActiveConsumers(final String
groupId,
final List<MemberDescription> members =
new
ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
if (!members.isEmpty()) {
- throw new IllegalStateException("Consumer group '" + groupId + "'
is still active "
- + "and has following members: " + members + ". "
- + "Make sure to stop all running application instances
before running the reset tool.");
+ if (options.has(forceOption)) {
+ System.out.println("Force deleting all active members in the
group: " + groupId);
+ try {
+ adminClient.removeMembersFromConsumerGroup(groupId, new
RemoveMembersFromConsumerGroupOptions()).all().get();
+ } catch (Exception e) {
+ throw e;
+ }
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId +
"' is still active "
+ + "and has following members: " + members + ". "
+ + "Make sure to stop all running application instances
before running the reset tool." +
+ "Try set '--force' in the cmdline to force delete
active members.");
Review comment:
```suggestion
+ " You can use option '--force' to remove active
members from the group.");
```
##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -186,9 +192,19 @@ private void validateNoActiveConsumers(final String
groupId,
final List<MemberDescription> members =
new
ArrayList<>(describeResult.describedGroups().get(groupId).get().members());
if (!members.isEmpty()) {
- throw new IllegalStateException("Consumer group '" + groupId + "'
is still active "
- + "and has following members: " + members + ". "
- + "Make sure to stop all running application instances
before running the reset tool.");
+ if (options.has(forceOption)) {
+ System.out.println("Force deleting all active members in the
group: " + groupId);
+ try {
+ adminClient.removeMembersFromConsumerGroup(groupId, new
RemoveMembersFromConsumerGroupOptions()).all().get();
+ } catch (Exception e) {
+ throw e;
+ }
+ } else {
+ throw new IllegalStateException("Consumer group '" + groupId +
"' is still active "
+ + "and has following members: " + members + ". "
+ + "Make sure to stop all running application instances
before running the reset tool." +
Review comment:
```suggestion
+ "Make sure to stop all running application
instances before running the reset tool."
```
##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -236,6 +252,8 @@ private void parseArguments(final String[] args) {
.withRequiredArg()
.ofType(String.class)
.describedAs("file name");
+ forceOption = optionParser.accepts("force", "Force remove members when
long session time out has been configured, " +
Review comment:
```suggestion
forceOption = optionParser.accepts("force", "Force the removal of
members of the consumer group. " +
```
##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -236,6 +252,8 @@ private void parseArguments(final String[] args) {
.withRequiredArg()
.ofType(String.class)
.describedAs("file name");
+ forceOption = optionParser.accepts("force", "Force remove members when
long session time out has been configured, " +
+ "please make sure to shut down all stream applications when
this option is specified to avoid unexpected rebalances.");
Review comment:
```suggestion
"Make sure to shut down all stream applications when this
option is specified to avoid unexpected rebalances.");
```
##########
File path: core/src/main/scala/kafka/tools/StreamsResetter.java
##########
@@ -236,6 +252,8 @@ private void parseArguments(final String[] args) {
.withRequiredArg()
.ofType(String.class)
.describedAs("file name");
+ forceOption = optionParser.accepts("force", "Force remove members when
long session time out has been configured, " +
Review comment:
```suggestion
forceOption = optionParser.accepts("force", "Force the removal of
members of the consumer group (intended to remove stopped members if a long
session timeout was used). " +
```
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java
##########
@@ -261,6 +261,42 @@ public void
shouldNotAllowToResetWhenIntermediateTopicAbsent() throws Exception
Assert.assertEquals(1, exitCode);
}
+ public void testResetWhenLongSessionTimeoutConfiguredWithForceOption()
throws Exception {
+ appID = testId + "-with-force-option";
+ streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+ streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "" +
STREAMS_CONSUMER_TIMEOUT * 100);
+
+ // Run
+ streams = new
KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ streams.start();
+ final List<KeyValue<Long, Long>> result =
IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig,
OUTPUT_TOPIC, 10);
+
+ streams.close();
+
+ // RESET
+ streams = new
KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ streams.cleanUp();
+
+ // Reset would fail since long session timeout has been configured
+ final boolean cleanResult = tryCleanGlobal(false, null, null);
+ Assert.assertEquals(false, cleanResult);
+
+ // Reset will success with --force, it will force delete active
members on broker side
+ cleanGlobal(false, "--force", null);
+
+ waitForEmptyConsumerGroup(adminClient, appID, TIMEOUT_MULTIPLIER *
CLEANUP_CONSUMER_TIMEOUT);
+
+ assertInternalTopicsGotDeleted(null);
+
+ // RE-RUN
Review comment:
Why do we need this part? Seems sufficient to end the test here?
##########
File path:
streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java
##########
@@ -261,6 +261,42 @@ public void
shouldNotAllowToResetWhenIntermediateTopicAbsent() throws Exception
Assert.assertEquals(1, exitCode);
}
+ public void testResetWhenLongSessionTimeoutConfiguredWithForceOption()
throws Exception {
+ appID = testId + "-with-force-option";
+ streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
+ streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "" +
STREAMS_CONSUMER_TIMEOUT * 100);
+
+ // Run
+ streams = new
KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ streams.start();
+ final List<KeyValue<Long, Long>> result =
IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig,
OUTPUT_TOPIC, 10);
+
+ streams.close();
+
+ // RESET
+ streams = new
KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
+ streams.cleanUp();
+
+ // Reset would fail since long session timeout has been configured
+ final boolean cleanResult = tryCleanGlobal(false, null, null);
+ Assert.assertEquals(false, cleanResult);
+
+ // Reset will success with --force, it will force delete active
members on broker side
+ cleanGlobal(false, "--force", null);
+
+ waitForEmptyConsumerGroup(adminClient, appID, TIMEOUT_MULTIPLIER *
CLEANUP_CONSUMER_TIMEOUT);
Review comment:
With `cleanGlobal` and `--force` the consumer group could be empty when
`cleanGlobal` returns, right? Hence, we should do this assertion without
timeout or retries?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3623,22 +3641,26 @@ public RemoveMembersFromConsumerGroupResult
removeMembersFromConsumerGroup(Strin
ConsumerGroupOperationContext<Map<MemberIdentity, Errors>,
RemoveMembersFromConsumerGroupOptions> context =
new ConsumerGroupOperationContext<>(groupId, options, deadline,
future);
- Call findCoordinatorCall = getFindCoordinatorCall(context,
- () -> getRemoveMembersFromGroupCall(context));
+ List<MemberIdentity> members;
+ if (options.removeAll()) {
+ members = getMembersFromGroup(groupId);
+ } else {
+ members =
options.members().stream().map(MemberToRemove::toMemberIdentity).collect(Collectors.toList());
+ }
+ Call findCoordinatorCall = getFindCoordinatorCall(context, () ->
getRemoveMembersFromGroupCall(context, members));
runnable.call(findCoordinatorCall, startFindCoordinatorMs);
return new RemoveMembersFromConsumerGroupResult(future,
options.members());
Review comment:
If `option.members()` is empty, it implies that we do a `removeAll()` --
hence, should we pass in `members` into the
`RemoveMembersFromConsumerGroupResult` instead of `options.members()` ?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3623,22 +3641,26 @@ public RemoveMembersFromConsumerGroupResult
removeMembersFromConsumerGroup(Strin
ConsumerGroupOperationContext<Map<MemberIdentity, Errors>,
RemoveMembersFromConsumerGroupOptions> context =
new ConsumerGroupOperationContext<>(groupId, options, deadline,
future);
- Call findCoordinatorCall = getFindCoordinatorCall(context,
- () -> getRemoveMembersFromGroupCall(context));
+ List<MemberIdentity> members;
+ if (options.removeAll()) {
+ members = getMembersFromGroup(groupId);
+ } else {
+ members =
options.members().stream().map(MemberToRemove::toMemberIdentity).collect(Collectors.toList());
+ }
+ Call findCoordinatorCall = getFindCoordinatorCall(context, () ->
getRemoveMembersFromGroupCall(context, members));
runnable.call(findCoordinatorCall, startFindCoordinatorMs);
return new RemoveMembersFromConsumerGroupResult(future,
options.members());
}
- private Call
getRemoveMembersFromGroupCall(ConsumerGroupOperationContext<Map<MemberIdentity,
Errors>, RemoveMembersFromConsumerGroupOptions> context) {
+ private Call
getRemoveMembersFromGroupCall(ConsumerGroupOperationContext<Map<MemberIdentity,
Errors>,
+ RemoveMembersFromConsumerGroupOptions> context,
List<MemberIdentity> members) {
Review comment:
nit: fix formatting:
```
private Call
getRemoveMembersFromGroupCall(ConsumerGroupOperationContext<Map<MemberIdentity,
Errors>, RemoveMembersFromConsumerGroupOptions> context,
List<MemberIdentity> members) {
```
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
##########
@@ -3660,7 +3686,7 @@ void handleResponse(AbstractResponse abstractResponse) {
// We set member.id to empty here explicitly, so that the
lookup will succeed as user doesn't
// know the exact member.id.
memberErrors.put(new MemberIdentity()
-
.setMemberId(JoinGroupRequest.UNKNOWN_MEMBER_ID)
+
.setMemberId(memberResponse.memberId())
Review comment:
Not sure if I understand the change. Also not sure if I can follow the
comments. Can you elaborate?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupOptions.java
##########
@@ -37,7 +38,15 @@ public
RemoveMembersFromConsumerGroupOptions(Collection<MemberToRemove> members)
this.members = new HashSet<>(members);
Review comment:
As we have different semantics for an empty collection (it was "remove
nothing" originally, and we change it to "remove all"), I am wondering if we
should do a check if `members` is empty or not and throw an exception if empty?
Or at least log a WARNING that empty implies "remove all" now?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.java
##########
@@ -51,9 +52,21 @@
if (throwable != null) {
result.completeExceptionally(throwable);
} else {
- for (MemberToRemove memberToRemove : memberInfos) {
- if (maybeCompleteExceptionally(memberErrors,
memberToRemove.toMemberIdentity(), result)) {
- return;
+ if (removeAll()) {
Review comment:
Not sure why the `removeAll()` case needs to be handled differently? Can
you elaboarte?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.java
##########
@@ -51,9 +52,21 @@
if (throwable != null) {
result.completeExceptionally(throwable);
} else {
- for (MemberToRemove memberToRemove : memberInfos) {
- if (maybeCompleteExceptionally(memberErrors,
memberToRemove.toMemberIdentity(), result)) {
- return;
+ if (removeAll()) {
Review comment:
Not sure why the `removeAll()` case needs to be handled differently? Can
you elaborate?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/admin/RemoveMembersFromConsumerGroupResult.java
##########
@@ -66,6 +79,9 @@
* Returns the selected member future.
*/
public KafkaFuture<Void> memberResult(MemberToRemove member) {
+ if (removeAll()) {
+ throw new IllegalArgumentException("The method: memberResult is
not applicable in 'removeAll' mode");
Review comment:
Why that? I understand that we expect that users don't know the memberId
if the so a "remove all"; however, I don't see why we need to disallow this
call? Can you elaborate?
##########
File path:
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
##########
@@ -379,6 +380,22 @@ private static MetadataResponse
prepareMetadataResponse(Cluster cluster, Errors
MetadataResponse.AUTHORIZED_OPERATIONS_OMITTED);
}
+ private static DescribeGroupsResponseData
prepareDescribeGroupsResponseData(String groupId, List<String> groupInstances,
Review comment:
Nit: formatting
```
private static DescribeGroupsResponseData
prepareDescribeGroupsResponseData(String groupId,
List<String> groupInstances,
List<TopicPartition> topicPartitions) {
```
##########
File path:
core/src/test/scala/integration/kafka/api/PlaintextAdminIntegrationTest.scala
##########
@@ -1017,47 +1017,70 @@ class PlaintextAdminIntegrationTest extends
BaseAdminIntegrationTest {
assertTrue(0 == list1.errors().get().size())
assertTrue(0 == list1.valid().get().size())
val testTopicName = "test_topic"
+ val testTopicName1 = testTopicName + "1"
+ val testTopicName2 = testTopicName + "2"
val testNumPartitions = 2
- client.createTopics(Collections.singleton(
- new NewTopic(testTopicName, testNumPartitions, 1.toShort))).all().get()
- waitForTopics(client, List(testTopicName), List())
+
+ client.createTopics(util.Arrays.asList(new NewTopic(testTopicName,
testNumPartitions, 1.toShort),
Review comment:
nit: formatting: move `new NewTopic(...)` to next line
----------------------------------------------------------------
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
