KAFKA-20524: CSV reset offset plan for kafka-share-groups.sh

[AndrewJSchofield]

This is the implementation of
https://cwiki.apache.org/confluence/display/KAFKA/KIP-1323%3A+Initialization+of+share+group+offsets+from+a+specific+offset+or+a+file.

Reviewers: Chia-Ping Tsai chia7712@gmail.com

[muralibasani]

This call resetPlanFromFile and next call resetFromFile seem to do same reading and parsing the entire csv. May be use opts to check for presence?

[muralibasani]

Is it better to interrupt the currentThread for RunTimeException?
For ExecutionException, getting the instance type for ex if it's of KafkaException(GroupAuthorizationException etc), it could be more visible to the caller ?

[muralibasani]

Not related to this pr, but this call sets offset to 0L, if opts.resetToOffsetOpt is empty or null. Is that ok instead of throwing an exception?
This is also the same in CosnumerGroupCommand too I think.

[AndrewJSchofield]

I think that's a tiny piece of defensive code in OffsetsUtils.resetToOffset. In practice, the command line argument spec for resetToOffsetOpt requires an argument of type Long. If no argument or a non-Long argument is specified with --to-offset, the command does fail.

[muralibasani]

@AndrewJSchofield thanks for the pr. I have some minor comments.

[AndrewJSchofield]

@AndrewJSchofield thanks for the pr. I have some minor comments.

Thanks for the review. I basically mapped the consumer group tool options across to the share group tool, and you're quite right that there are a few things to improve. I'll sort it out.

[chia7712]

@AndrewJSchofield Out of curiosity, does the KIP's motivation include aligning offset-related options? If so, should we also support the shift-by option?

[chia7712]

To define the scope use --all-topics or --topic. One scope must be specified unless you use '--from-file'

It looks like we should remove the condition if (!options.has(topicOpt) && !options.has(allTopicsOpt)) from line 226, right?

[chia7712]

I understand this implementation aligns with resetToCurrent, but it feels a bit old-school to me. We could simplify it using the following style

    public Map<TopicPartition, OffsetAndMetadata> resetToCurrentForShareGroup(Collection<TopicPartition> partitionsToReset, Map<TopicPartition, SharePartitionOffsetInfo> currentOffsetInfo) {
        var partitioned = partitionsToReset.stream().collect(Collectors.partitioningBy(currentOffsetInfo::containsKey));
        var partitionsToResetWithStartOffset = partitioned.get(true);
        var partitionsToResetWithoutStartOffset = partitioned.get(false);

        var preparedOffsetsForPartitionsWithStartOffset = partitionsToResetWithStartOffset.stream()
            .collect(Collectors.toMap(Function.identity(), topicPartition -> new OffsetAndMetadata(currentOffsetInfo.get(topicPartition).startOffset())));

        getLogEndOffsets(partitionsToResetWithoutStartOffset).forEach((tp, logOffsetResult) -> {
            if (!(logOffsetResult instanceof OffsetsUtils.LogOffset logOffset)) {
                throw new IllegalStateException("Error getting ending offset of topic partition: " + tp);
            }
            preparedOffsetsForPartitionsWithStartOffset.put(tp, new OffsetAndMetadata(logOffset.value));
        });

        return preparedOffsetsForPartitionsWithStartOffset;
    }

[chia7712]

BTW, I'm fine with addressing this in a separate PR

[AndrewJSchofield]

This KIP is still in voting and I'm getting the PR ready in anticipation. Entirely happy to iterate on it.

[AndrewJSchofield]

@AndrewJSchofield Out of curiosity, does the KIP's motivation include aligning offset-related options? If so, should we also support the shift-by option?

@chia7712 The motivation was really to enable offsets to be exported from one group and used to initialise another. I'm not really averse to shift-by, but hadn't planned to do it. There is also another difference and that is that you can only reset the offsets on one share group at a time. When resetting offsets for consumer groups, you can list multiple groups or reset all groups at once.