suppress java.nio.file.NoSuchFileException when renaming deleted segm…

…ents (#419)

NoSuchFileException is-a IOException and Kafka handles it similar to a failure of the disk device and marks the log dir as offline due to a previous IO exception. However, this behavior is overly pessimistic and should be suppressed.

core/src/main/scala/kafka/log/Log.scala

@@ -2682,7 +2682,15 @@ object Log extends Logging {
                                       logDirFailureChannel: LogDirFailureChannel,
                                       producerStateManager: ProducerStateManager,
                                       logPrefix: String): Unit = {
-    segmentsToDelete.foreach(_.changeFileSuffixes("", Log.DeletedFileSuffix))
+    segmentsToDelete.foreach(logSegment => {
+      try {
+        logSegment.changeFileSuffixes("", Log.DeletedFileSuffix)
+      } catch {
+        case e: java.nio.file.NoSuchFileException => {
+          info(s"Failed to rename log file ${logSegment.log.file.getAbsoluteFile} with ${Log.DeletedFileSuffix} suffix because it was already deleted")
+        }
+      }
+    })
     val snapshotsToDelete = if (deleteProducerStateSnapshots)
       segmentsToDelete.flatMap { segment =>
         producerStateManager.removeAndMarkSnapshotForDeletion(segment.baseOffset)}