MINOR: Remove unnecessary Optional from offsetsToSnapshot (#19613)

Reviewers: PoAn Yang <[email protected]>, Ken Huang
 <[email protected]>, TengYao Chi <[email protected]>, Chia-Ping Tsai
 <[email protected]>

Author: Yunyung

storage/src/main/java/org/apache/kafka/storage/internals/log/UnifiedLog.java

@@ -2416,15 +2416,13 @@ public static void rebuildProducerState(ProducerStateManager producerStateManage
                                             Time time,
                                             boolean reloadFromCleanShutdown,
                                             String logPrefix) throws IOException {
-        List<Optional<Long>> offsetsToSnapshot = new ArrayList<>();
-        if (segments.nonEmpty()) {
-            long lastSegmentBaseOffset = segments.lastSegment().get().baseOffset();
-            Optional<LogSegment> lowerSegment = segments.lowerSegment(lastSegmentBaseOffset);
-            Optional<Long> nextLatestSegmentBaseOffset = lowerSegment.map(LogSegment::baseOffset);
-            offsetsToSnapshot.add(nextLatestSegmentBaseOffset);
-            offsetsToSnapshot.add(Optional.of(lastSegmentBaseOffset));
-        }
-        offsetsToSnapshot.add(Optional.of(lastOffset));
+        List<Long> offsetsToSnapshot = new ArrayList<>();
+        segments.lastSegment().ifPresent(lastSegment -> {
+            long lastSegmentBaseOffset = lastSegment.baseOffset();
+            segments.lowerSegment(lastSegmentBaseOffset).ifPresent(s -> offsetsToSnapshot.add(s.baseOffset()));
+            offsetsToSnapshot.add(lastSegmentBaseOffset);
+        });
+        offsetsToSnapshot.add(lastOffset);
 
         LOG.info("{}Loading producer state till offset {}", logPrefix, lastOffset);
 
@@ -2443,11 +2441,9 @@ public static void rebuildProducerState(ProducerStateManager producerStateManage
             // To avoid an expensive scan through all the segments, we take empty snapshots from the start of the
             // last two segments and the last offset. This should avoid the full scan in the case that the log needs
             // truncation.
-            for (Optional<Long> offset : offsetsToSnapshot) {
-                if (offset.isPresent()) {
-                    producerStateManager.updateMapEndOffset(offset.get());
-                    producerStateManager.takeSnapshot();
-                }
+            for (long offset : offsetsToSnapshot) {
+                producerStateManager.updateMapEndOffset(offset);
+                producerStateManager.takeSnapshot();
             }
         } else {
             LOG.info("{}Reloading from producer snapshot and rebuilding producer state from offset {}", logPrefix, lastOffset);
@@ -2469,7 +2465,7 @@ public static void rebuildProducerState(ProducerStateManager producerStateManage
                     long startOffset = Utils.max(segment.baseOffset(), producerStateManager.mapEndOffset(), logStartOffset);
                     producerStateManager.updateMapEndOffset(startOffset);
 
-                    if (offsetsToSnapshot.contains(Optional.of(segment.baseOffset()))) {
+                    if (offsetsToSnapshot.contains(segment.baseOffset())) {
                         producerStateManager.takeSnapshot();
                     }
                     int maxPosition = segment.size();