Status
Current state: Under Discussion vote
Discussion thread: https://lists.apache.org/thread.html/r2dc1cabfb75df6117bc464c29f0700dd1214fd1352bacfa99381b9bf%40%3Cdev.kafka.apache.org%3E
JIRA:
Motivation
- user has to specify input topic name in the stream reset tool in order to purge its offsets
Adding a new option which could automatically detect all external topics and reset their corresponding offsets
Public Interfaces
- A new resetAllExternalTopicsOption will be added for user to cleanup all externally committed offsets
- `--input-topic`and `--intermediate-topic` flag using regular expressions
add reset-all-external-topics option
public class StreamsResetter { private static OptionSpecBuilder resetAllExternalTopicsOption; private void parseArguments(final String[] args) { ... resetAllExternalTopicsOption = optionParser.accepts("reset-all-external-topics", "Reset tool such that when enabled, delete offsets for all involved topics"); } private boolean isInputTopic(final String topic) { final List<String> inputTopics = options.valuesOf(inputTopicsOption); return inputTopics.stream().anyMatch(inputTopic -> { final Pattern pattern = Pattern.compile(inputTopic); return inputTopic.equals(topic) || pattern.matcher(topic).matches(); }); } private boolean isIntermediateTopic(final String topic) { final List<String> intermediateTopics = options.valuesOf(intermediateTopicsOption); return intermediateTopics.stream().anyMatch(intermediateTopic -> { final Pattern pattern = Pattern.compile(intermediateTopic); return intermediateTopic.equals(topic) || pattern.matcher(topic).matches(); }); } }
Compatibility, Deprecation, and Migration Plan
Because it is an option addition Compatibility, Deprecation, and Migration Plan are not a problem
Rejected Alternatives
Not applicable.