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

Code before:
                final String dirName = taskDirectory.file().getName();
                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 Tasks 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());
                    final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager(
                            id,
                            eosEnabled,
                            logContext,
                            this,
                            subTopology.storeToChangelogTopic(),
                            inputPartitions,
                            stateUpdaterEnabled
                    );
                    final StartupContext initContext = new StartupContext(id, config, stateManager);
                    // TODO: we need to pass a proper logPrefix
                    StateManagerUtil.registerStateStores(log, "", subTopology, stateManager, this, initContext);
                    for (final StateStore stateStore : subTopology.stateStores()) {
                        if (!stateStore.isOpen()) {
                            throw new IllegalStateException("StateStore [" + stateStore.name() + "] is not open");
                        }
                    }
                    tasksForLocalState.put(id, new StartupState(id, subTopology, stateManager));
                }
            }
        }
    }

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

    public StartupState removeStartupTask(final TaskId taskId) {
        final StartupState task = tasksForLocalState.remove(taskId);
        if (task != null) {
            lockedTasksToOwner.replace(taskId, Thread.currentThread());
        }
        return task;
    }

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


Review comment (bug):
I assume, the rest of stores will be updated later? right now there is only one Store updated: RockDBStore. I opened a random store(InMemoryKeyValueStore) and it sets the open field to true in the init method, so it will start failing here, I think

Code after:

    /**
     * @throws LockException If another process already locks any of StateDirectory
     * @throws ProcessorStateException if any of task directory does not exist and could not be created
    */
    public void initializeStartupStores(final TopologyMetadata topologyMetadata,
                                        final LogContext logContext,
                                        final StreamsMetricsImpl metricsImpl) {
        final List<TaskDirectory> nonEmptyTaskDirectories = listNonEmptyTaskDirectories();
        if (hasPersistentStores && !nonEmptyTaskDirectories.isEmpty()) {
            final boolean eosEnabled = StreamsConfigUtils.eosEnabled(config);

            // Initialize thread-specific resources needed to open stores in the state directory
            final String threadLogPrefix = String.format("[%s]", Thread.currentThread().getName());
            final ThreadCache dummyCache = new ThreadCache(new LogContext(threadLogPrefix), 0L, metricsImpl);

            // 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