gardnervickers commented on a change in pull request #10896:
URL: https://github.com/apache/kafka/pull/10896#discussion_r656296678


##########
File path: core/src/main/scala/kafka/log/ProducerStateManager.scala
##########
@@ -828,15 +834,50 @@ class ProducerStateManager(val topicPartition: 
TopicPartition,
    * Removes the producer state snapshot file metadata corresponding to the 
provided offset if it exists from this
    * ProducerStateManager, and deletes the backing snapshot file.
    */
-  private[log] def removeAndDeleteSnapshot(snapshotOffset: Long): Unit = {
+  private def removeAndDeleteSnapshot(snapshotOffset: Long): Unit = {
     Option(snapshots.remove(snapshotOffset)).foreach(_.deleteIfExists())
   }
+
+  /**
+   * Removes the producer state snapshot file metadata corresponding to the 
provided offset if it exists from this
+   * ProducerStateManager, and renames the backing snapshot file to have the 
Log.DeletionSuffix.
+   *
+   * Note: This method is safe to use with async deletes. If a race occurs and 
the snapshot file
+   *       is deleted without this ProducerStateManager instance knowing, the 
resulting exception on
+   *       SnapshotFile rename will be ignored and None will be returned.
+   */
+  private[log] def removeAndMarkSnapshotForDeletion(snapshotOffset: Long): 
Option[SnapshotFile] = {
+    Option(snapshots.remove(snapshotOffset)).flatMap { snapshot => {
+      // If the file cannot be renamed, it likely means that the file was 
deleted already.
+      // This can happen due to the way we construct an intermediate producer 
state manager
+      // during log recovery, and use it to issue deletions prior to creating 
the "real"
+      // producer state manager.
+      //
+      // In any case, removeAndMarkSnapshotForDeletion is intended to be used 
for snapshot file
+      // deletion, so ignoring the exception here just means that the intended 
operation was
+      // already completed.
+      try {
+        snapshot.renameTo(Log.DeletedFileSuffix)
+      } catch {
+        case _: NoSuchFileException =>
+          return None

Review comment:
       Sure we can do that.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


Reply via email to