Repository: apache/kafka
File: shell/src/main/java/org/apache/kafka/shell/MetadataShell.java (Java)

Code before:
     * TOCTOU bug here where someone could create and lock the lockfile in between our check
     * and our use. However, this is very unlikely to ever be a problem in practice, and closing
     * this hole would require the parent directory to always be writable when loading a
     * snapshot so that we could create our .lock file there.
     */
    static FileLock takeDirectoryLockIfExists(File directory) throws IOException {
        if (new File(directory, ".lock").exists()) {
            return takeDirectoryLock(directory);
        } else {
            return null;
        }
    }

    /**
     * Take the FileLock in the given directory.
     */
    static FileLock takeDirectoryLock(File directory) throws IOException {
        FileLock fileLock = new FileLock(new File(directory, ".lock"));
        try {
            if (!fileLock.tryLock()) {
                throw new RuntimeException("Unable to lock " + directory.getAbsolutePath() +
                    ". Please ensure that no broker or controller process is using this " +
                    "directory before proceeding.");
            }
        } catch (Throwable e) {
            fileLock.destroy();
            throw e;
        }
        return fileLock;
    }

    private final MetadataShellState state;

    private final String snapshotPath;

    private final FaultHandler faultHandler;

    private final MetadataShellPublisher publisher;

    private FileLock fileLock;

    private SnapshotFileReader snapshotFileReader;

    private MetadataLoader loader;

    public MetadataShell(
        String snapshotPath,
        FaultHandler faultHandler
    ) {
        this.state = new MetadataShellState();
        this.snapshotPath = snapshotPath;

Review comment (bug):
I think the problem with this code is that RuntimeException is a Throwable so the previous code used to delete the file even though grabbing the lock failed. In general the JVM should delete the lock file only if it was able to acquired the lock. The guarantee is that only the process holding the lock is able to delete the file.

With this change the metadata shell never deletes the lock file even if it is the last process to hold the lock.

Code after:
     * TOCTOU bug here where someone could create and lock the lockfile in between our check
     * and our use. However, this is very unlikely to ever be a problem in practice, and closing
     * this hole would require the parent directory to always be writable when loading a
     * snapshot so that we could create our .lock file there.
     */
    static FileLock takeDirectoryLockIfExists(File directory) throws IOException {
        if (new File(directory, ".lock").exists()) {
            return takeDirectoryLock(directory);
        } else {
            return null;
        }
    }

    /**
     * Take the FileLock in the given directory.
     */
    static FileLock takeDirectoryLock(File directory) throws IOException {
        FileLock fileLock = new FileLock(new File(directory, ".lock"));
        try {
            if (!fileLock.tryLock()) {
                throw new RuntimeException("Unable to lock " + directory.getAbsolutePath() +
                    ". Please ensure that no broker or controller process is using this " +
                    "directory before proceeding.");
            }
        } catch (Throwable e) {
            fileLock.unlockAndClose();
            throw e;
        }
        return fileLock;
    }

    private final MetadataShellState state;

    private final String snapshotPath;

    private final FaultHandler faultHandler;

    private final MetadataShellPublisher publisher;

    private FileLock fileLock;

    private SnapshotFileReader snapshotFileReader;

    private MetadataLoader loader;

    public MetadataShell(
        String snapshotPath,
        FaultHandler faultHandler
    ) {
        this.state = new MetadataShellState();
        this.snapshotPath = snapshotPath;