KAFKA-19244: Add support for kafka-streams-groups.sh options (offset-related APIs) [1/N] #19646
Conversation
mjsax
added
streams
KIP-1071
There was a problem hiding this comment.
Thanks for the PR @aliehsaeedii I made a pass over the non-testing code, I'll do more in a follow up review
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java
Show resolved
Hide resolved
tools/src/main/java/org/apache/kafka/tools/streams/StreamsGroupCommand.java
Outdated
Show resolved
Hide resolved
| throw new IllegalArgumentException("Cannot shift offset for partition " + topicPartition + " since there is no current committed offset"); | ||
| } | ||
|
|
||
| return currentOffset.offset() + shiftBy; |
There was a problem hiding this comment.
Do we want to do any validation on what the acceptable values for shiftBy?
There was a problem hiding this comment.
I assume any long value should be accepted. Good point. I'll add some tests.
There was a problem hiding this comment.
Seems shiftby 150 and -150 are already existing.
aliehsaeedii
changed the title
lucasbru
changed the title
There was a problem hiding this comment.
Hey @aliehsaeedii ! Thanks for the PR. Mostly looking good to me, left a few comments
|
|
||
| private Collection<TopicPartition> getPartitionsToReset(String groupId) throws ExecutionException, InterruptedException { | ||
| if (opts.options.has(opts.allTopicsOpt)) { | ||
| return getCommittedOffsets(groupId).keySet(); |
There was a problem hiding this comment.
I think this may not be the correct behavior. We only want to reset the offset of source topics, never the offset of repartition topics (which are also committed). For a clean reset, as implemented by the streams resetter, I think we want to delete the internal topics (through the other options). Resetting repartition topics to the beginning seems incorrect, since it will mean that we will reconsume these records in the downstream subtopologies.
@bbejeck Wdyt?
We are already describing the streams group, so I think this shouldn't be too hard to implement - we have to pass the source topics into this function.
| return null; | ||
| } | ||
|
|
||
| Optional<Map<String, Map<TopicPartition, OffsetAndMetadata>>> resetPlanFromFile() { |
There was a problem hiding this comment.
I assume quite a bit of this code was copied from the consumer group tool. Is there any chance of sharing some code? May not be the case, but since most of this code is just dealing with topic partitions and not with tasks, I wonder if there is an easy way to refactor functions like this into a common utility.
| } | ||
| } catch (IOException e) { | ||
| e.printStackTrace(); | ||
| // Ignore. |
There was a problem hiding this comment.
Don't we want to fail here somehow?
This PR implements all the options for
--reset-offset, (supports--execute,--dry-run,--export). Includes unit and integrationtests.
duration
datetime