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

Code before:
     */
    void recycle() {
        log.debug("Recycling state for {} task {}.", taskType, taskId);

        if (!stateUpdaterEnabled) {
            final List<TopicPartition> allChangelogs = getAllChangelogTopicPartitions();
            changelogReader.unregister(allChangelogs);
        }
    }

    void transitionTaskType(final TaskType newType) {
        if (taskType.equals(newType)) {
            throw new IllegalStateException("Tried to recycle state for task type conversion but new type was the same.");
        }

        taskType = newType;
    }

    @Override
    public void updateChangelogOffsets(final Map<TopicPartition, Long> writtenOffsets) {
        for (final Map.Entry<TopicPartition, Long> entry : writtenOffsets.entrySet()) {
            final StateStoreMetadata store = findStore(entry.getKey());

            if (store != null) {
                store.setOffset(entry.getValue());

                log.debug("State store {} updated to written offset {} at changelog {}",
                        store.stateStore.name(), store.offset, store.changelogPartition);
            }
        }
    }

    @Override
    public void checkpoint() {
        // checkpoint those stores that are only logged and persistent to the checkpoint file
        final Map<TopicPartition, Long> checkpointingOffsets = new HashMap<>();
        for (final StateStoreMetadata storeMetadata : stores.values()) {
            if (storeMetadata.commitCallback != null && !storeMetadata.corrupted) {
                try {
                    storeMetadata.commitCallback.onCommit();
                } catch (final IOException e) {
                    throw new ProcessorStateException(
                            format("%sException caught while trying to checkpoint store, " +
                                    "changelog partition %s", logPrefix, storeMetadata.changelogPartition),
                            e
                    );
                }
            }

            // store is logged, persistent, not corrupted, and has a valid current offset
            if (storeMetadata.changelogPartition != null &&

Review comment (suggestion):
This is a piggy-backed fix: I realized that when we recycle tasks, the log context should also be updated to change the prefix. Otherwise it would be very confusing in the logs since a recycled `standby` task would still print as `active` task in the prefix and vice versa.

Code after:
     */
    void recycle() {
        log.debug("Recycling state for {} task {}.", taskType, taskId);

        if (!stateUpdaterEnabled) {
            final List<TopicPartition> allChangelogs = getAllChangelogTopicPartitions();
            changelogReader.unregister(allChangelogs);
        }

        // when the state manager is recycled to be used, future writes may bypass its store's caching
        // layer if they are from restoration, hence we need to clear the state store's caches just in case
        // See KAFKA-14172 for details
        if (!stores.isEmpty()) {
            log.debug("Clearing all store caches registered in the state manager: {}", stores);
            for (final StateStoreMetadata metadata : stores.values()) {
                final StateStore store = metadata.stateStore;

                if (store instanceof CachedStateStore) {
                    ((CachedStateStore) store).clearCache();
                }
                log.trace("Cleared cache {}", store.name());
            }
        }
    }

    void transitionTaskType(final TaskType newType, final LogContext logContext) {
        if (taskType.equals(newType)) {
            throw new IllegalStateException("Tried to recycle state for task type conversion but new type was the same.");
        }

        taskType = newType;
        log = logContext.logger(ProcessorStateManager.class);
    }

    @Override
    public void updateChangelogOffsets(final Map<TopicPartition, Long> writtenOffsets) {
        for (final Map.Entry<TopicPartition, Long> entry : writtenOffsets.entrySet()) {
            final StateStoreMetadata store = findStore(entry.getKey());

            if (store != null) {
                store.setOffset(entry.getValue());

                log.debug("State store {} updated to written offset {} at changelog {}",
                        store.stateStore.name(), store.offset, store.changelogPartition);
            }
        }
    }

    @Override
    public void checkpoint() {
        // checkpoint those stores that are only logged and persistent to the checkpoint file