Repository: apache/kafka
File: core/src/main/scala/kafka/log/LogManager.scala (Scala)

Code before:
    destLog.removeLogMetrics()
    if (updateHighWatermark && sourceLog.isDefined) {
      destLog.updateHighWatermark(sourceLog.get.highWatermark)
    }

    // Now that future replica has been successfully renamed to be the current replica
    // Update the cached map and log cleaner as appropriate.
    futureLogs.remove(topicPartition)
    currentLogs.put(topicPartition, destLog)
    if (cleaner != null) {
      sourceLog.foreach { srcLog =>
        cleaner.alterCheckpointDir(topicPartition, srcLog.parentDirFile, destLog.parentDirFile)
      }
      resumeCleaning(topicPartition)
    }

    try {
      sourceLog.foreach { srcLog =>
        srcLog.renameDir(UnifiedLog.logDeleteDirName(topicPartition), true)
        // Now that replica in source log directory has been successfully renamed for deletion.
        // Close the log, update checkpoint files, and enqueue this log to be deleted.
        srcLog.close()
        val logDir = srcLog.parentDirFile
        val logsToCheckpoint = logsInDir(logDir)
        checkpointRecoveryOffsetsInDir(logDir, logsToCheckpoint)
        checkpointLogStartOffsetsInDir(logDir, logsToCheckpoint)
        srcLog.removeLogMetrics()
        addLogToBeDeleted(srcLog)
      }
      destLog.newMetrics()
    } catch {
      case e: KafkaStorageException =>
        // If sourceLog's log directory is offline, we need close its handlers here.
        // handleLogDirFailure() will not close handlers of sourceLog because it has been removed from currentLogs map
        sourceLog.foreach { srcLog =>
          srcLog.closeHandlers()
          srcLog.removeLogMetrics()
        }
        throw e
    }
  }

  /**
    * Rename the directory of the given topic-partition "logdir" as "logdir.uuid.delete" and
    * add it in the queue for deletion.
    *
    * @param topicPartition TopicPartition that needs to be deleted
    * @param isFuture True iff the future log of the specified partition should be deleted
    * @param checkpoint True if checkpoints must be written
    * @return the removed log
    */

Review comment (suggestion):
>         //The log will be deleted asynchronously by the background delete-logs thread.
>         // File handles are intentionally left open; Unix semantics allow the renamed files
>         // to remain accessible until all handles are closed.

How about the following?

File handles are intentionally left open; Unix semantics allow the renamed files
to remain accessible until all handles are closed. 
The log will be deleted asynchronously by the background delete-logs thread.
File handles are closed and files are deleted after a configured delay log.segment.delete.delay.ms.
At that time, the expectation is that no other concurrent operations need to access
the deleted file handles any more.

Code after:
    destLog.removeLogMetrics()
    if (updateHighWatermark && sourceLog.isDefined) {
      destLog.updateHighWatermark(sourceLog.get.highWatermark)
    }

    // Now that future replica has been successfully renamed to be the current replica
    // Update the cached map and log cleaner as appropriate.
    futureLogs.remove(topicPartition)
    currentLogs.put(topicPartition, destLog)
    if (cleaner != null) {
      sourceLog.foreach { srcLog =>
        cleaner.alterCheckpointDir(topicPartition, srcLog.parentDirFile, destLog.parentDirFile)
      }
      resumeCleaning(topicPartition)
    }

    try {
      sourceLog.foreach { srcLog =>
        srcLog.renameDir(UnifiedLog.logDeleteDirName(topicPartition), true)
        // Now that replica in source log directory has been successfully renamed for deletion,
        // update checkpoint files and enqueue this log to be deleted.
        // Note: We intentionally do NOT close the log here to avoid race conditions where concurrent
        // operations (e.g., log flusher, fetch requests) might encounter ClosedChannelException.
        // The log will be deleted asynchronously by the background delete-logs thread.
        // File handles are intentionally left open; Unix semantics allow the renamed files
        // to remain accessible until all handles are closed.
        val logDir = srcLog.parentDirFile
        val logsToCheckpoint = logsInDir(logDir)
        checkpointRecoveryOffsetsInDir(logDir, logsToCheckpoint)
        checkpointLogStartOffsetsInDir(logDir, logsToCheckpoint)
        srcLog.removeLogMetrics()
        addLogToBeDeleted(srcLog)
      }
      destLog.newMetrics()
    } catch {
      case e: KafkaStorageException =>
        // If sourceLog's log directory is offline, we need close its handlers here.
        // handleLogDirFailure() will not close handlers of sourceLog because it has been removed from currentLogs map
        sourceLog.foreach { srcLog =>
          srcLog.closeHandlers()
          srcLog.removeLogMetrics()
        }
        throw e
    }
  }

  /**
    * Rename the directory of the given topic-partition "logdir" as "logdir.uuid.delete" and
    * add it in the queue for deletion.
    *
    * @param topicPartition TopicPartition that needs to be deleted