Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java (Java)

Code before:
            lockedTasksToOwner.replace(taskId, Thread.currentThread());
        }
        return task;
    }

    public void closeStartupTasks() {
        closeStartupTasks(t -> true);
    }

    private void closeStartupTasks(final Predicate<Task> predicate) {
        if (!tasksForLocalState.isEmpty()) {
            // "drain" Tasks first to ensure that we don't try to close Tasks that another thread is attempting to close
            final Set<Task> drainedTasks = new HashSet<>(tasksForLocalState.size());
            for (final Map.Entry<TaskId, Task> entry : tasksForLocalState.entrySet()) {
                if (predicate.test(entry.getValue()) && tasksForLocalState.remove(entry.getKey()) != null) {
                    // only add to our list of drained Tasks if we exclusively "claimed" a Task from tasksForLocalState
                    // to ensure we don't accidentally try to drain the same Task multiple times from concurrent threads
                    drainedTasks.add(entry.getValue());
                }
            }

            // now that we have exclusive ownership of the drained tasks, close them
            for (final Task task : drainedTasks) {
                // main thread locked the task initially on startup, but has moved on and will not unlock
                // so we need to explicitly swap lock ownership here as this method is called by a StreamThread
                lockedTasksToOwner.replace(task.id(), Thread.currentThread());
                task.suspend();
                task.closeClean();
            }
        }
    }

    public UUID initializeProcessId() {
        if (!hasPersistentStores) {
            final UUID processId = UUID.randomUUID();
            log.info("Created new process id: {}", processId);
            return processId;
        }

        if (!lockStateDirectory()) {
            log.error("Unable to obtain lock as state directory is already locked by another process");
            throw new StreamsException(String.format("Unable to initialize state, this can happen if multiple instances of " +
                                           "Kafka Streams are running in the same state directory " +
                                           "(current state directory is [%s]", stateDir.getAbsolutePath()));
        }

        final File processFile = new File(stateDir, PROCESS_FILE_NAME);
        final ObjectMapper mapper = new ObjectMapper();

        try {
            if (processFile.exists()) {

Review comment (suggestion):
This is the fix.  Once a `StreamThread` receives its assignment, it will close the startup tasks.  But during the closing process, the `StandbyTask.closeClean()` method will eventually call the `StatemanagerUtil.closeStateManager` method which needs to lock the state directory, but locking requires the calling thread be the current owner of the state directory.  Since the main thread grabs the lock on startup but moves on without releasing it, we need to update ownership explicitly here in order for the stream thread to close the startup task and begin processing. 

Code after:
            lockedTasksToOwner.replace(taskId, Thread.currentThread());
        }
        return task;
    }

    public void closeStartupTasks() {
        closeStartupTasks(t -> true);
    }

    private void closeStartupTasks(final Predicate<Task> predicate) {
        if (!tasksForLocalState.isEmpty()) {
            // "drain" Tasks first to ensure that we don't try to close Tasks that another thread is attempting to close
            final Set<Task> drainedTasks = new HashSet<>(tasksForLocalState.size());
            for (final Map.Entry<TaskId, Task> entry : tasksForLocalState.entrySet()) {
                if (predicate.test(entry.getValue()) && removeStartupTask(entry.getKey()) != null) {
                    // only add to our list of drained Tasks if we exclusively "claimed" a Task from tasksForLocalState
                    // to ensure we don't accidentally try to drain the same Task multiple times from concurrent threads
                    drainedTasks.add(entry.getValue());
                }
            }

            // now that we have exclusive ownership of the drained tasks, close them
            for (final Task task : drainedTasks) {
                task.suspend();
                task.closeClean();
            }
        }
    }

    public UUID initializeProcessId() {
        if (!hasPersistentStores) {
            final UUID processId = UUID.randomUUID();
            log.info("Created new process id: {}", processId);
            return processId;
        }

        if (!lockStateDirectory()) {
            log.error("Unable to obtain lock as state directory is already locked by another process");
            throw new StreamsException(String.format("Unable to initialize state, this can happen if multiple instances of " +
                                           "Kafka Streams are running in the same state directory " +
                                           "(current state directory is [%s]", stateDir.getAbsolutePath()));
        }

        final File processFile = new File(stateDir, PROCESS_FILE_NAME);
        final ObjectMapper mapper = new ObjectMapper();

        try {
            if (processFile.exists()) {
                try {
                    final StateDirectoryProcessFile processFileData = mapper.readValue(processFile, StateDirectoryProcessFile.class);
                    log.info("Reading UUID from process file: {}", processFileData.processId);