<commit_before>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.streams.processor.internals;

import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.processor.TaskId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.channels.OverlappingFileLockException;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.HashMap;

/**
 * Manages the directories where the state of Tasks owned by a {@link StreamThread} are
 * stored. Handles creation/locking/unlocking/cleaning of the Task Directories. This class is not
 * thread-safe.
 */
public class StateDirectory {

    static final String LOCK_FILE_NAME = ".lock";
    private static final Logger log = LoggerFactory.getLogger(StateDirectory.class);

    private final File stateDir;
    private final HashMap<TaskId, FileChannel> channels = new HashMap<>();
    private final HashMap<TaskId, FileLock> locks = new HashMap<>();

    public StateDirectory(final String applicationId, final String stateDirConfig) {
        final File baseDir = new File(stateDirConfig);
        if (!baseDir.exists() && !baseDir.mkdirs()) {
            throw new ProcessorStateException(String.format("state directory [%s] doesn't exist and couldn't be created",
                                                            stateDirConfig));
        }
        stateDir = new File(baseDir, applicationId);
        if (!stateDir.exists() && !stateDir.mkdir()) {
            throw new ProcessorStateException(String.format("state directory [%s] doesn't exist and couldn't be created",
                                                            stateDir.getPath()));
        }

    }

    /**
     * Get or create the directory for the {@link TaskId}
     * @param taskId
     * @return directory for the {@link TaskId}
     */
    public File directoryForTask(final TaskId taskId) {
        final File taskDir = new File(stateDir, taskId.toString());
        if (!taskDir.exists() && !taskDir.mkdir()) {
            throw new ProcessorStateException(String.format("task directory [%s] doesn't exist and couldn't be created",
                                                            taskDir.getPath()));
        }
        return taskDir;
    }

    /**
     * Get the lock for the {@link TaskId}s directory if it is available
     * @param taskId
     * @param retry
     * @return true if successful
     * @throws IOException
     */
    public boolean lock(final TaskId taskId, int retry) throws IOException {
        // we already have the lock so bail out here
        if (locks.containsKey(taskId)) {
            return true;
        }
        final File lockFile = new File(directoryForTask(taskId), LOCK_FILE_NAME);
        final FileChannel channel = getOrCreateFileChannel(taskId, lockFile.toPath());

        FileLock lock = tryAcquireLock(channel);
        while (lock == null && retry > 0) {
            try {
                Thread.sleep(200);
            } catch (Exception ex) {
                // do nothing
            }
            retry--;
            lock = tryAcquireLock(channel);
        }
        if (lock != null) {
            locks.put(taskId, lock);
        }
        return lock != null;
    }

    /**
     * Unlock the state directory for the given {@link TaskId}
     * @param taskId
     * @throws IOException
     */
    public void unlock(final TaskId taskId) throws IOException {
        final FileLock lock = locks.remove(taskId);
        if (lock != null) {
            lock.release();
            final FileChannel fileChannel = channels.remove(taskId);
            if (fileChannel != null) {
                fileChannel.close();
            }
        }<commit_msg>KAFKA-4392; Handle NoSuchFileException gracefully in StateDirectory<commit_after>/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.kafka.streams.processor.internals;

import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.processor.TaskId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.nio.channels.FileChannel;
import java.nio.channels.FileLock;
import java.nio.channels.OverlappingFileLockException;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.HashMap;

/**
 * Manages the directories where the state of Tasks owned by a {@link StreamThread} are
 * stored. Handles creation/locking/unlocking/cleaning of the Task Directories. This class is not
 * thread-safe.
 */
public class StateDirectory {

    static final String LOCK_FILE_NAME = ".lock";
    private static final Logger log = LoggerFactory.getLogger(StateDirectory.class);

    private final File stateDir;
    private final HashMap<TaskId, FileChannel> channels = new HashMap<>();
    private final HashMap<TaskId, FileLock> locks = new HashMap<>();

    public StateDirectory(final String applicationId, final String stateDirConfig) {
        final File baseDir = new File(stateDirConfig);
        if (!baseDir.exists() && !baseDir.mkdirs()) {
            throw new ProcessorStateException(String.format("state directory [%s] doesn't exist and couldn't be created",
                                                            stateDirConfig));
        }
        stateDir = new File(baseDir, applicationId);
        if (!stateDir.exists() && !stateDir.mkdir()) {
            throw new ProcessorStateException(String.format("state directory [%s] doesn't exist and couldn't be created",
                                                            stateDir.getPath()));
        }

    }

    /**
     * Get or create the directory for the {@link TaskId}
     * @param taskId
     * @return directory for the {@link TaskId}
     */
    public File directoryForTask(final TaskId taskId) {
        final File taskDir = new File(stateDir, taskId.toString());
        if (!taskDir.exists() && !taskDir.mkdir()) {
            throw new ProcessorStateException(String.format("task directory [%s] doesn't exist and couldn't be created",
                                                            taskDir.getPath()));
        }
        return taskDir;
    }

    /**
     * Get the lock for the {@link TaskId}s directory if it is available
     * @param taskId
     * @param retry
     * @return true if successful
     * @throws IOException
     */
    public boolean lock(final TaskId taskId, int retry) throws IOException {
        // we already have the lock so bail out here
        if (locks.containsKey(taskId)) {
            return true;
        }
        final File lockFile = new File(directoryForTask(taskId), LOCK_FILE_NAME);

        final FileChannel channel;

        try {
            channel = getOrCreateFileChannel(taskId, lockFile.toPath());
        } catch (NoSuchFileException e) {
            // FileChannel.open(..) could throw NoSuchFileException when there is another thread
            // concurrently deleting the parent directory (i.e. the directory of the taskId) of the lock
            // file, in this case we will return immediately indicating locking failed.
            return false;
        }

        FileLock lock = tryAcquireLock(channel);
        while (lock == null && retry > 0) {
            try {
                Thread.sleep(200);
            } catch (Exception ex) {
                // do nothing
            }
            retry--;
            lock = tryAcquireLock(channel);
        }
        if (lock != null) {
            locks.put(taskId, lock);
        }
        return lock != null;
    }

    /**
     * Unlock the state directory for the given {@link TaskId}
     * @param taskId
     * @throws IOException
     */
    public void unlock(final TaskId taskId) throws IOException {
        final FileLock lock = locks.remove(taskId);
        if (lock != null) {
            lock.release();
            final FileChannel fileChannel = channels.remove(taskId);
            if (fileChannel != null) {
                fileChannel.close();
            }
        }
{'fraction_non_alphanumeric': 0.09133111312947825, 'fraction_numerical': 0.0024220405691795337, 'mean_word_length': 2.450557103064067, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6651801', 'n_tokens_mistral': 2549, 'n_tokens_neox': 2488, 'n_words': 870}