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

Code before:
                final TaskId id = parseTaskDirectoryName(dirName, taskDirectory.namedTopology());
                final ProcessorTopology subTopology = topologyMetadata.buildSubtopology(id);

                // we still check if the task's sub-topology is stateful, even though we know its directory contains state,
                // because it's possible that the topology has changed since that data was written, and is now stateless
                // this therefore prevents us from creating unnecessary stores just because of some left-over state
                if (subTopology.hasStateWithChangelogs()) {
                    final Set<TopicPartition> inputPartitions = topologyMetadata.nodeToSourceTopics(id).values().stream()
                            .flatMap(Collection::stream)
                            .map(t -> new TopicPartition(t, id.partition()))
                            .collect(Collectors.toSet());
                    // Open a temporary state manager that will open the stores inside the subtopology
                    final ProcessorStateManager temporaryStateManager = ProcessorStateManager.createStartupTaskStateManager(
                        id,
                        eosEnabled,
                        logContext,
                        this,
                        subTopology.storeToChangelogTopic(),
                        inputPartitions
                    );

                    final StartupContext initContext = new StartupContext(id, config, temporaryStateManager, metricsImpl, cache);
                    try {
                        StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
                    } catch (final TaskCorruptedException tce) {
                        log.warn("Failed to register startup state stores for task {}: {}", id, tce.getMessage());
                    } finally {
                        // Make sure the state manager writes the local checkpoint file before closing the stores
                        // This will be replaced in the future when removing the checkpoint file dependency.
                        temporaryStateManager.checkpoint();
                        temporaryStateManager.close();
                    }
                    tasksForLocalState.put(id, new StartupState(id, subTopology, temporaryStateManager));
                }
            }
        }
    }

    public boolean hasStartupTasks() {
        return !tasksForLocalState.isEmpty();
    }

    public boolean removeStartupState(final TaskId taskId) {
        final StartupState startupState = tasksForLocalState.remove(taskId);
        final boolean removed = startupState != null;
        if (removed) {
            lockedTasksToOwner.replace(taskId, Thread.currentThread());
        }
        return removed;
    }


Review comment (suggestion):
`registerStateStores` also can throw a `StreamsException`, `LockException` , or  a `ProcessorStateException` should we handle these exceptions as well? If not add some comments on the reasoning

Code after:
            // discover all non-empty task directories in StateDirectory
            for (final TaskDirectory taskDirectory : nonEmptyTaskDirectories) {
                final String dirName = taskDirectory.file().getName();
                final TaskId task = parseTaskDirectoryName(dirName, taskDirectory.namedTopology());
                final ProcessorTopology subTopology = topologyMetadata.buildSubtopology(task);

                // we still check if the task's sub-topology is stateful, even though we know its directory contains state,
                // because it's possible that the topology has changed since that data was written, and is now stateless
                // this therefore prevents us from creating unnecessary stores just because of some left-over state
                if (subTopology.hasStateWithChangelogs()) {
                    final Set<TopicPartition> inputPartitions = topologyMetadata.nodeToSourceTopics(task).values().stream()
                            .flatMap(Collection::stream)
                            .map(t -> new TopicPartition(t, task.partition()))
                            .collect(Collectors.toSet());
                    // Open a temporary state manager that will open the stores inside the subtopology
                    final ProcessorStateManager temporaryStateManager = ProcessorStateManager.createStartupTaskStateManager(
                        task,
                        eosEnabled,
                        logContext,
                        this,
                        subTopology.storeToChangelogTopic(),
                        inputPartitions
                    );

                    final StartupContext initContext = new StartupContext(task, config, temporaryStateManager, metricsImpl, dummyCache);
                    try {
                        // We only handle TaskCorruptedException at this point. Any other exception is considered fatal.
                        StateManagerUtil.registerStateStores(log, threadLogPrefix, subTopology, temporaryStateManager, this, initContext);
                        temporaryStateManager.checkpoint();
                    } catch (final TaskCorruptedException tce) {
                        // At this point, we only log a warning and continue with the startup store initialization.
                        // The task-corrupted exception will be handled in the first Task assignment phase.
                        log.warn("Failed to register startup state stores for task {}: {}", task, tce.getMessage());
                    } finally {
                        // Make sure the state manager writes the local checkpoint file before closing the stores
                        // This will be replaced in the future when removing the checkpoint file dependency.
                        temporaryStateManager.close();
                    }
                    tasksInLocalState.add(task);
                }
            }
        }
    }

    public boolean hasStartupTasks() {
        return !tasksInLocalState.isEmpty();
    }

    public synchronized boolean removeStartupState(final TaskId taskId) {
        final boolean removed = tasksInLocalState.remove(taskId);
        if (removed) {