class ProcessorStateManager {

    public void close(Map<TopicPartition, Long> ackedOffsets) throws IOException {
        try {
            // attempting to flush and close the stores, just in case they
            // are not closed by a ProcessorNode yet
            if (!stores.isEmpty()) {
                log.debug("Closing stores.");
                for (Map.Entry<String, StateStore> entry : stores.entrySet()) {
                    log.debug("Closing storage engine {}", entry.getKey());
                    entry.getValue().flush();
                    entry.getValue().close();
                }

                Map<TopicPartition, Long> checkpointOffsets = new HashMap<>();
                for (String storeName : stores.keySet()) {
                    TopicPartition part;
                    if (loggingEnabled.contains(storeName))
                        part = new TopicPartition(storeChangelogTopic(applicationId, storeName), getPartition(storeName));
                    else
                        part = new TopicPartition(storeName, getPartition(storeName));

                    // only checkpoint the offset to the offsets file if it is persistent;
                    if (stores.get(storeName).persistent()) {
                        Long offset = ackedOffsets.get(part);

                        if (offset != null) {
                            // store the last offset + 1 (the log position after restoration)
                            checkpointOffsets.put(part, offset + 1);
                        } else {
                            // if no record was produced. we need to check the restored offset.
                            offset = restoredOffsets.get(part);
                            if (offset != null)
                                checkpointOffsets.put(part, offset);
                        }
                    }
                }

                // write the checkpoint file before closing, to indicate clean shutdown
                OffsetCheckpoint checkpoint = new OffsetCheckpoint(new File(this.baseDir, CHECKPOINT_FILE_NAME));
                checkpoint.write(checkpointOffsets);
            }
        } finally {
            // release the state directory directoryLock
            directoryLock.release();
            directoryLock.channel().close();
        }
    }

    public  ProcessorStateManager(String applicationId, int defaultPartition, Collection<TopicPartition> sources, File baseDir, Consumer<byte[], byte[]> restoreConsumer, boolean isStandby);

}