From 67196e00964fefb2d9ef9b06f95c140efdd4777c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Armando=20Garc=C3=ADa=20Sancio?= Date: Wed, 17 Mar 2021 16:23:52 -0700 Subject: [PATCH 1/3] MINOR: Remove use of NoSuchElementException --- .../scala/kafka/raft/KafkaMetadataLog.scala | 39 +++++++------------ 1 file changed, 15 insertions(+), 24 deletions(-) diff --git a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala index 037e3c91db165..77d59c1da9c8d 100644 --- a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala +++ b/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, ()) } } } From 2f1f185168b71ab185d7f1363ab8c244a1547ad0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Armando=20Garc=C3=ADa=20Sancio?= Date: Wed, 17 Mar 2021 17:26:07 -0700 Subject: [PATCH 2/3] Find the earliest snapshot once when validating offset and epoch --- .../java/org/apache/kafka/raft/ReplicatedLog.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java b/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java index 9a4233aa75330..d385e0b948f5e 100644 --- a/raft/src/main/java/org/apache/kafka/raft/ReplicatedLog.java +++ b/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 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 From b9fa75752bba61a2f6c76d24a3234168963d6e77 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Armando=20Garc=C3=ADa=20Sancio?= Date: Thu, 18 Mar 2021 08:27:25 -0700 Subject: [PATCH 3/3] Use ascending and descending iterators --- .../scala/kafka/raft/KafkaMetadataLog.scala | 37 ++++++++++++------- 1 file changed, 23 insertions(+), 14 deletions(-) diff --git a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala index 77d59c1da9c8d..fc7e465c21b48 100644 --- a/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala +++ b/core/src/main/scala/kafka/raft/KafkaMetadataLog.scala @@ -18,7 +18,7 @@ package kafka.raft import java.io.{File, IOException} import java.nio.file.{Files, NoSuchFileException} -import java.util.concurrent.ConcurrentSkipListMap +import java.util.concurrent.ConcurrentSkipListSet import java.util.{Optional, Properties} import kafka.api.ApiVersion @@ -37,9 +37,8 @@ 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. Using a Map instead of a Set so that there is no - // need to handle NoSuchElementException. - snapshotIds: ConcurrentSkipListMap[OffsetAndEpoch, Unit], + // polling thread when snapshots are created. + snapshotIds: ConcurrentSkipListSet[OffsetAndEpoch], topicPartition: TopicPartition, maxFetchSizeInBytes: Int, val fileDeleteDelayMs: Long // Visible for testing, @@ -237,7 +236,7 @@ final class KafkaMetadataLog private ( override def readSnapshot(snapshotId: OffsetAndEpoch): Optional[RawSnapshotReader] = { try { - if (snapshotIds.containsKey(snapshotId)) { + if (snapshotIds.contains(snapshotId)) { Optional.of(FileRawSnapshotReader.open(log.dir.toPath, snapshotId)) } else { Optional.empty() @@ -249,20 +248,30 @@ final class KafkaMetadataLog private ( } override def latestSnapshotId(): Optional[OffsetAndEpoch] = { - Optional.ofNullable(snapshotIds.lastEntry).map(_.getKey) + val descending = snapshotIds.descendingIterator + if (descending.hasNext) { + Optional.of(descending.next) + } else { + Optional.empty() + } } override def earliestSnapshotId(): Optional[OffsetAndEpoch] = { - Optional.ofNullable(snapshotIds.firstEntry).map(_.getKey) + val ascendingIterator = snapshotIds.iterator + if (ascendingIterator.hasNext) { + Optional.of(ascendingIterator.next) + } else { + Optional.empty() + } } override def onSnapshotFrozen(snapshotId: OffsetAndEpoch): Unit = { - snapshotIds.put(snapshotId, ()) + snapshotIds.add(snapshotId) } override def deleteBeforeSnapshot(logStartSnapshotId: OffsetAndEpoch): Boolean = { latestSnapshotId().asScala match { - case Some(snapshotId) if (snapshotIds.containsKey(logStartSnapshotId) && + case Some(snapshotId) if (snapshotIds.contains(logStartSnapshotId) && startOffset < logStartSnapshotId.offset && logStartSnapshotId.offset <= snapshotId.offset && log.maybeIncrementLogStartOffset(logStartSnapshotId.offset, SnapshotGenerated)) => @@ -281,9 +290,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.headMap(logStartSnapshotId, false).entrySet.iterator + val expiredSnapshotIdsIter = snapshotIds.headSet(logStartSnapshotId, false).iterator while (expiredSnapshotIdsIter.hasNext) { - val snapshotId = expiredSnapshotIdsIter.next().getKey + val snapshotId = expiredSnapshotIdsIter.next() // 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() @@ -357,8 +366,8 @@ object KafkaMetadataLog { private def recoverSnapshots( log: Log - ): ConcurrentSkipListMap[OffsetAndEpoch, Unit] = { - val snapshotIds = new ConcurrentSkipListMap[OffsetAndEpoch, Unit]() + ): ConcurrentSkipListSet[OffsetAndEpoch] = { + val snapshotIds = new ConcurrentSkipListSet[OffsetAndEpoch]() // Scan the log directory; deleting partial snapshots and older snapshot, only remembering immutable snapshots start // from logStartOffset Files @@ -378,7 +387,7 @@ object KafkaMetadataLog { // Delete partial snapshot, deleted snapshot and older snapshot Files.deleteIfExists(snapshotPath.path) } else { - snapshotIds.put(snapshotPath.snapshotId, ()) + snapshotIds.add(snapshotPath.snapshotId) } } }