MINOR: Remove use of NoSuchElementException

core/src/main/scala/kafka/raft/KafkaMetadataLog.scala

@@ -18,8 +18,8 @@ package kafka.raft
 
 import java.io.{File, IOException}
 import java.nio.file.{Files, NoSuchFileException}
-import java.util.concurrent.ConcurrentSkipListSet
-import java.util.{NoSuchElementException, Optional, Properties}
+import java.util.concurrent.ConcurrentSkipListMap
+import java.util.{Optional, Properties}
 
 import kafka.api.ApiVersion
 import kafka.log.{AppendOrigin, Log, LogConfig, LogOffsetSnapshot, SnapshotGenerated}
@@ -37,8 +37,9 @@ final class KafkaMetadataLog private (
   log: Log,
   scheduler: Scheduler,
   // This object needs to be thread-safe because it is used by the snapshotting thread to notify the
-  // polling thread when snapshots are created.
-  snapshotIds: ConcurrentSkipListSet[OffsetAndEpoch],
+  // polling thread when snapshots are created. Using a Map instead of a Set so that there is no
+  // need to handle NoSuchElementException.
+  snapshotIds: ConcurrentSkipListMap[OffsetAndEpoch, Unit],
   topicPartition: TopicPartition,
   maxFetchSizeInBytes: Int,
   val fileDeleteDelayMs: Long // Visible for testing,
@@ -236,7 +237,7 @@ final class KafkaMetadataLog private (
 
   override def readSnapshot(snapshotId: OffsetAndEpoch): Optional[RawSnapshotReader] = {
     try {
-      if (snapshotIds.contains(snapshotId)) {
+      if (snapshotIds.containsKey(snapshotId)) {
         Optional.of(FileRawSnapshotReader.open(log.dir.toPath, snapshotId))
       } else {
         Optional.empty()
@@ -248,30 +249,20 @@ final class KafkaMetadataLog private (
   }
 
   override def latestSnapshotId(): Optional[OffsetAndEpoch] = {
-    try {
-      Optional.of(snapshotIds.last)
-    } catch {
-      case _: NoSuchElementException =>
-        Optional.empty()
-    }
+    Optional.ofNullable(snapshotIds.lastEntry).map(_.getKey)
   }
 
   override def earliestSnapshotId(): Optional[OffsetAndEpoch] = {
-    try {
-      Optional.of(snapshotIds.first)
-    } catch {
-      case _: NoSuchElementException =>
-        Optional.empty()
-    }
+      Optional.ofNullable(snapshotIds.firstEntry).map(_.getKey)
   }
 
   override def onSnapshotFrozen(snapshotId: OffsetAndEpoch): Unit = {
-    snapshotIds.add(snapshotId)
+    snapshotIds.put(snapshotId, ())
   }
 
   override def deleteBeforeSnapshot(logStartSnapshotId: OffsetAndEpoch): Boolean = {
     latestSnapshotId().asScala match {
-      case Some(snapshotId) if (snapshotIds.contains(logStartSnapshotId) &&
+      case Some(snapshotId) if (snapshotIds.containsKey(logStartSnapshotId) &&
         startOffset < logStartSnapshotId.offset &&
         logStartSnapshotId.offset <= snapshotId.offset &&
         log.maybeIncrementLogStartOffset(logStartSnapshotId.offset, SnapshotGenerated)) =>
@@ -290,9 +281,9 @@ final class KafkaMetadataLog private (
    * Removes all snapshots on the log directory whose epoch and end offset is less than the giving epoch and end offset.
    */
   private def removeSnapshotFilesBefore(logStartSnapshotId: OffsetAndEpoch): Unit = {
-    val expiredSnapshotIdsIter = snapshotIds.headSet(logStartSnapshotId, false).iterator()
+    val expiredSnapshotIdsIter = snapshotIds.headMap(logStartSnapshotId, false).entrySet.iterator
     while (expiredSnapshotIdsIter.hasNext) {
-      val snapshotId = expiredSnapshotIdsIter.next()
+      val snapshotId = expiredSnapshotIdsIter.next().getKey
       // If snapshotIds contains a snapshot id, the KafkaRaftClient and Listener can expect that the snapshot exists
       // on the file system, so we should first remove snapshotId and then delete snapshot file.
       expiredSnapshotIdsIter.remove()
@@ -366,8 +357,8 @@ object KafkaMetadataLog {
 
   private def recoverSnapshots(
     log: Log
-  ): ConcurrentSkipListSet[OffsetAndEpoch] = {
-    val snapshotIds = new ConcurrentSkipListSet[OffsetAndEpoch]()
+  ): ConcurrentSkipListMap[OffsetAndEpoch, Unit] = {
+    val snapshotIds = new ConcurrentSkipListMap[OffsetAndEpoch, Unit]()
     // Scan the log directory; deleting partial snapshots and older snapshot, only remembering immutable snapshots start
     // from logStartOffset
     Files
@@ -387,7 +378,7 @@ object KafkaMetadataLog {
             // Delete partial snapshot, deleted snapshot and older snapshot
             Files.deleteIfExists(snapshotPath.path)
           } else {
-            snapshotIds.add(snapshotPath.snapshotId)
+            snapshotIds.put(snapshotPath.snapshotId, ())
           }
         }
       }

raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java

@@ -84,11 +84,13 @@ public interface ReplicatedLog extends Closeable {
     default ValidOffsetAndEpoch validateOffsetAndEpoch(long offset, int epoch) {
         if (startOffset() == 0 && offset == 0) {
             return ValidOffsetAndEpoch.valid(new OffsetAndEpoch(0, 0));
-        } else if (
-                earliestSnapshotId().isPresent() &&
-                ((offset < startOffset()) ||
-                 (offset == startOffset() && epoch != earliestSnapshotId().get().epoch) ||
-                 (epoch < earliestSnapshotId().get().epoch))
+        }
+
+        Optional<OffsetAndEpoch> earliestSnapshotId = earliestSnapshotId();
+        if (earliestSnapshotId.isPresent() &&
+            ((offset < startOffset()) ||
+             (offset == startOffset() && epoch != earliestSnapshotId.get().epoch) ||
+             (epoch < earliestSnapshotId.get().epoch))
         ) {
             /* Send a snapshot if the leader has a snapshot at the log start offset and
              * 1. the fetch offset is less than the log start offset or