Repository: apache/kafka
File: tools/src/main/java/org/apache/kafka/tools/consumer/group/ShareGroupCommand.java (Java)

Code before:
            topics.forEach(topic -> {
                try {
                    deleteResult.topicResult(topic).get();
                    topicLevelResult.put(topic, null);
                } catch (ExecutionException | InterruptedException e) {
                    topicLevelResult.put(topic, e.getCause());
                }
            });

            return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
        }

        void resetOffsets() {
            String groupId = opts.options.valueOf(opts.groupOpt);
            try {
                ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
                if (!(GroupState.EMPTY.equals(shareGroupDescription.groupState()) || GroupState.DEAD.equals(shareGroupDescription.groupState()))) {
                    CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
                }
                resetOffsetsForInactiveGroup(groupId);
            } catch (InterruptedException ie) {
                throw new RuntimeException(ie);
            } catch (ExecutionException ee) {
                Throwable cause = ee.getCause();
                if (cause instanceof GroupIdNotFoundException) {
                    resetOffsetsForInactiveGroup(groupId);
                } else if (cause instanceof KafkaException) {
                    CommandLineUtils.printErrorAndExit(cause.getMessage());
                } else {
                    throw new RuntimeException(cause);
                }
            }
        }

        private void resetOffsetsForInactiveGroup(String groupId) {
            try {
                Collection<TopicPartition> partitionsToReset = getPartitionsToReset(groupId);
                Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(groupId, partitionsToReset);
                boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
                if (!dryRun) {
                    adminClient.alterShareGroupOffsets(groupId,
                        offsetsToReset.entrySet().stream()
                            .collect(Collectors.toMap(
                                Entry::getKey, entry -> entry.getValue().offset()
                            ))
                    ).all().get();
                }
                OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
            } catch (InterruptedException ie) {
                throw new RuntimeException(ie);
            } catch (ExecutionException ee) {

Review comment (suggestion):
Just wondering, why is there a retry mechanism when `GroupIdNotFoundException` is thrown? Is this for better reliability? I see ConsumerGroupCommand also does this.

Code after:
            topics.forEach(topic -> {
                try {
                    deleteResult.topicResult(topic).get();
                    topicLevelResult.put(topic, null);
                } catch (ExecutionException | InterruptedException e) {
                    topicLevelResult.put(topic, e.getCause());
                }
            });

            return new SimpleImmutableEntry<>(topLevelException, topicLevelResult);
        }

        void resetOffsets() {
            String groupId = opts.options.valueOf(opts.groupOpt);
            try {
                ShareGroupDescription shareGroupDescription = describeShareGroups(List.of(groupId)).get(groupId);
                if (!(GroupState.EMPTY.equals(shareGroupDescription.groupState()) || GroupState.DEAD.equals(shareGroupDescription.groupState()))) {
                    CommandLineUtils.printErrorAndExit(String.format("Share group '%s' is not empty.", groupId));
                }
                resetOffsetsForInactiveGroup(groupId);
            } catch (InterruptedException ie) {
                throw new RuntimeException(ie);
            } catch (ExecutionException ee) {
                Throwable cause = ee.getCause();
                if (cause instanceof GroupIdNotFoundException) {
                    resetOffsetsForInactiveGroup(groupId);
                } else if (cause instanceof KafkaException) {
                    CommandLineUtils.printErrorAndExit(cause.getMessage());
                } else {
                    throw new RuntimeException(cause);
                }
            }
        }

        private void resetOffsetsForInactiveGroup(String groupId) {
            try {
                Collection<TopicPartition> partitionsToReset = getPartitionsToReset(groupId);
                Map<TopicPartition, OffsetAndMetadata> offsetsToReset = prepareOffsetsToReset(partitionsToReset);
                boolean dryRun = opts.options.has(opts.dryRunOpt) || !opts.options.has(opts.executeOpt);
                if (!dryRun) {
                    adminClient.alterShareGroupOffsets(groupId,
                        offsetsToReset.entrySet().stream()
                            .collect(Collectors.toMap(
                                Entry::getKey, entry -> entry.getValue().offset()
                            ))
                    ).all().get();
                }
                OffsetsUtils.printOffsetsToReset(Map.of(groupId, offsetsToReset));
            } catch (InterruptedException ie) {
                throw new RuntimeException(ie);
            } catch (ExecutionException ee) {