Status
Current state: Under DiscussionAdopted for 2.6
Vote thread: [Vote] KIP-571: Add option to force remove members in StreamsResetter
Discussion thread: [Discuss] KIP-571: Add option to force remove members in StreamsResetter
...
Sometimes people want to reset the stream application sooner , but blocked by the left-over members inside the group coordinator, which only expire expires after session timeout. When the user configures a really long session timeout, it could prevent the group from clearing. We should consider adding the support to cleanup clean up members by forcing them to leave the group. To do that,
- If the stream application is already on static membership, we could call directly from adminClient.removeMembersFromGroup
If the application is on dynamic membership, we should modify adminClient.removeMembersFromGroup interface to allow deletion based on member.id.
Proposed Changes && Public Interfaces
KIP-569 plan to support the ability to force removing members in `StreamsResetter`, this basically involves 2 public interfaces changes as below:
1) Changes in org.apache.kafka.clients.admin.MemberToRemove
Changes includes:
- New field:
- private String memberId;
- New methods:
- public MemberToRemove() {}
- public MemberToRemove setGroupInstanceId(String groupInstanceId) {}
- public MemberToRemove setMemberId(String memberId) {}
- Deprecated methods:
- public MemberToRemove(String groupInstanceId) {}
Previously the org.apache.kafka.clients.admin.KafkaAdminClient#removeMembersFromConsumerGroup only support to remove members by specifying the groupInstanceId, which means it only support remove static members, so the memberId is added to support removing dynamic members .
...
2) Add cmdline arg force-remove-member to StreamsResetter
The current StreamsResetter will throw exception when found active members:
Code Block |
---|
bin/kafka-streams-application-reset.sh --application-id streams-wordcount --input-topics streams-plaintext-input |
While with the new arg, the StreamsResetter will force removing all active members :
...
we could enhance KafkaAdminClient#removeMembersFromConsumerGroup to support remove all members(static&dynamic) in a certain group.
Public Interfaces
Client side changes:
kafkaAdminClient related changes:
Code Block | ||
---|---|---|
| ||
public class RemoveMembersFromConsumerGroupOptions {
// newly added method
public boolean removeAll() {}
// newly added constructor
public RemoveMembersFromConsumerGroupOptions() {}
} |
CmdLine API change:
Code Block | ||
---|---|---|
| ||
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") |
Proposed Changes
KIP-571 plan to support the ability to force remove members in StreamsResetter, this involves public interfaces changes as below:
1) Changes related to kafkaAdminclient
The old constructor RemoveMembersFromConsumerGroupOptions(Collection<MemberToRemove>members) will imply the non removeAll or remove-specific members scenario and it will throw IllegalArgumentException if empty members is provided.
When the newly added constructor RemoveMembersFromConsumerGroupOptions() is used, it implies the removeAll scenario, underlyingly it set field members to an empty set and removeAll() will return true, in this case, KafkaAdminClient#removeMembersFromConsumerGroup will remove all members in the given group, it will first query the members of the given group and then issue a LeaveGroupRequest with all members specified. Accordingly, the RemoveMembersFromConsumerGroupResult also adds a new private method removeAll() to imply different handling in RemoveMembersFromConsumerGroupResult#all,memberResult. .Under the removeAll scenario, RemoveMembersFromConsumerGroupResult#memberInfos should be empty, the memberResult() is not applicable, the all() should check if RemoveMembersFromConsumerGroupResult#future 100% succeed, if not, it will throw the first exception captured.
2) Add cmdline option --force to StreamsResetter
While with the new option, the StreamsResetter will force remove all active members by calling KafkaAdminClient#removeMembersFromConsumerGroup with removeAll specified. All the deleted members' info will be logged out if all removals succeed. Otherwise, the first member removal error will be thrown.
The detailed member removal error cases could be found in KIP-345, which introduced static membership and batch removal on the broker side.
Broker side logic has no change, GroupCoordinator#handleLeaveGroup can handle both dynamic and static member removals since 2.4.
Operational steps of the StreamsResetter:
- Stop all the consumers (of streams instances), and wait until the shutdown is complete, otherwise, unexpected rebalance will be triggered.
- Use StreamsResetter with --force in case the stopped consumers are still registered at the broker side, without --force, it will fail with an exception suggesting using --force.
Compatibility, Deprecation, and Migration Plan
- Because no classes/method will be removed but only deprecated, this change 1) will be fully backward compatible
- The new added cmdline arg forceoption -remove-member force is also backward compatible because it's a new feature, if not specified, the StreamsRestterStreamsResetter's behaviour remains unchanged.s behavior remains unchanged.
The requirement to use this feature:
Release version requirement | Expected error if requirement not satisfied | |
---|---|---|
broker side | >= 2.4 | UnsupportedVersionException will be thrown because the batch removal feature was introduced since version 2.4 |
client side | update to this KIP | "force is not a recognized option" will be in the client side log |
Rejected Alternatives
None
...