MINOR: Convert GroupOffsetsResetterOptions to record (apache#22251)

Convert `GroupOffsetsResetterOptions` to record.

Reviewers: Ken Huang <s7133700@gmail.com>, Maros Orsak
<maros.orsak159@gmail.com>, PoAn Yang <payang@apache.org>

Author: majialoong

tools/src/main/java/org/apache/kafka/tools/GroupOffsetsResetter.java

@@ -94,9 +94,9 @@ public static void printOffsetsToReset(Map<String, Map<TopicPartition, OffsetAnd
     }
 
     public Optional<Map<String, Map<TopicPartition, OffsetAndMetadata>>> resetPlanFromFile() {
-        if (opts.resetFromFileOpt != null && !opts.resetFromFileOpt.isEmpty()) {
+        if (opts.resetFromFileOpt() != null && !opts.resetFromFileOpt().isEmpty()) {
             try {
-                String resetPlanPath = opts.resetFromFileOpt.get(0);
+                String resetPlanPath = opts.resetFromFileOpt().get(0);
                 String resetPlanCsv = Utils.readFileAsString(resetPlanPath);
                 Map<String, Map<TopicPartition, OffsetAndMetadata>> resetPlan = parseResetPlan(resetPlanCsv);
                 return Optional.of(resetPlan);
@@ -109,7 +109,7 @@ public Optional<Map<String, Map<TopicPartition, OffsetAndMetadata>>> resetPlanFr
     private Map<String, Map<TopicPartition, OffsetAndMetadata>> parseResetPlan(String resetPlanCsv) {
         ObjectReader csvReader = CsvUtils.readerFor(CsvUtils.CsvRecordNoGroup.class);
         String[] lines = resetPlanCsv.split("\n");
-        boolean isSingleGroupQuery = opts.groupOpt.size() == 1;
+        boolean isSingleGroupQuery = opts.groupOpt().size() == 1;
         boolean isOldCsvFormat = false;
         try {
             if (lines.length > 0) {
@@ -126,7 +126,7 @@ private Map<String, Map<TopicPartition, OffsetAndMetadata>> parseResetPlan(Strin
         try {
             // Single group CSV format: "topic,partition,offset"
             if (isSingleGroupQuery && isOldCsvFormat) {
-                String group = opts.groupOpt.get(0);
+                String group = opts.groupOpt().get(0);
                 for (String line : lines) {
                     CsvUtils.CsvRecordNoGroup rec = csvReader.readValue(line, CsvUtils.CsvRecordNoGroup.class);
                     dataMap.computeIfAbsent(group, k -> new HashMap<>())
@@ -298,8 +298,8 @@ public Stream<TopicPartition> parseTopicsWithPartitions(String topicArg) {
     }
 
     public Map<TopicPartition, OffsetAndMetadata> resetToOffset(Collection<TopicPartition> partitionsToReset) {
-        long offset = opts.resetToOffsetOpt != null && !opts.resetToOffsetOpt.isEmpty()
-            ? opts.resetToOffsetOpt.get(0)
+        long offset = opts.resetToOffsetOpt() != null && !opts.resetToOffsetOpt().isEmpty()
+            ? opts.resetToOffsetOpt().get(0)
             : 0L;
         return checkOffsetsRange(partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), tp -> offset)))
             .entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> new OffsetAndMetadata(e.getValue())));
@@ -336,7 +336,7 @@ public Map<TopicPartition, OffsetAndMetadata> resetByShiftBy(
         Map<TopicPartition, OffsetAndMetadata> currentCommittedOffsets) {
 
         Map<TopicPartition, Long> requestedOffsets = partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), topicPartition -> {
-            long shiftBy = opts.resetShiftByOpt;
+            long shiftBy = opts.resetShiftByOpt();
             OffsetAndMetadata currentOffset = currentCommittedOffsets.get(topicPartition);
 
             if (currentOffset == null) {
@@ -351,7 +351,7 @@ public Map<TopicPartition, OffsetAndMetadata> resetByShiftBy(
 
     public Map<TopicPartition, OffsetAndMetadata> resetToDateTime(Collection<TopicPartition> partitionsToReset) {
         try {
-            long timestamp = Utils.getDateTime(opts.resetToDatetimeOpt.get(0));
+            long timestamp = Utils.getDateTime(opts.resetToDatetimeOpt().get(0));
             Map<TopicPartition, LogOffsetResult> logTimestampOffsets =
                 getLogTimestampOffsets(partitionsToReset, timestamp);
             return partitionsToReset.stream().collect(Collectors.toMap(Function.identity(), topicPartition -> {
@@ -367,7 +367,7 @@ public Map<TopicPartition, OffsetAndMetadata> resetToDateTime(Collection<TopicPa
     }
 
     public Map<TopicPartition, OffsetAndMetadata> resetByDuration(Collection<TopicPartition> partitionsToReset) {
-        String duration = opts.resetByDurationOpt;
+        String duration = opts.resetByDurationOpt();
         Duration durationParsed = Duration.parse(duration);
         Instant now = Instant.now();
         durationParsed.negated().addTo(now);
@@ -489,7 +489,7 @@ public List<TopicPartition> filterNonExistentPartitions(Collection<TopicPartitio
     }
 
     private <T extends AbstractOptions<T>> T withTimeoutMs(T options) {
-        int t = (int) opts.timeoutMsOpt;
+        int t = (int) opts.timeoutMsOpt();
         return options.timeoutMs(t);
     }
 
@@ -508,41 +508,20 @@ public static class Unknown implements LogOffsetResult { }
     public static class Ignore implements LogOffsetResult { }
 
 
-    public static class GroupOffsetsResetterOptions {
-        List<String> groupOpt;
-        List<Long> resetToOffsetOpt;
-        List<String> resetFromFileOpt;
-        List<String> resetToDatetimeOpt;
-        String resetByDurationOpt;
-        Long resetShiftByOpt;
-        long timeoutMsOpt;
+    public record GroupOffsetsResetterOptions(
+        List<String> groupOpt,
+        List<Long> resetToOffsetOpt,
+        List<String> resetFromFileOpt,
+        List<String> resetToDatetimeOpt,
+        String resetByDurationOpt,
+        Long resetShiftByOpt,
+        long timeoutMsOpt) {
 
         public GroupOffsetsResetterOptions(
             List<String> groupOpt,
-            List<Long> resetToOffsetOpt,
-            List<String> resetFromFileOpt,
             List<String> resetToDatetimeOpt,
-            String resetByDurationOpt,
-            Long resetShiftByOpt,
             long timeoutMsOpt) {
-
-            this.groupOpt = groupOpt;
-            this.resetToOffsetOpt = resetToOffsetOpt;
-            this.resetFromFileOpt = resetFromFileOpt;
-            this.resetToDatetimeOpt = resetToDatetimeOpt;
-            this.resetByDurationOpt = resetByDurationOpt;
-            this.resetShiftByOpt = resetShiftByOpt;
-            this.timeoutMsOpt = timeoutMsOpt;
-        }
-
-        public GroupOffsetsResetterOptions(
-            List<String> groupOpt,
-            List<String> resetToDatetimeOpt,
-            long timeoutMsOpt) {
-
-            this.groupOpt = groupOpt;
-            this.resetToDatetimeOpt = resetToDatetimeOpt;
-            this.timeoutMsOpt = timeoutMsOpt;
+            this(groupOpt, null, null, resetToDatetimeOpt, null, null, timeoutMsOpt);
         }
     }
 }