### Question:
ProcessorStateManager implements StateManager { @Override public void close(final Map<TopicPartition, Long> ackedOffsets) throws ProcessorStateException { RuntimeException firstException = null; try { if (!stores.isEmpty()) { log.debug("{} Closing its state manager and all the registered state stores", logPrefix); for (final Map.Entry<String, StateStore> entry : stores.entrySet()) { log.debug("{} Closing storage engine {}", logPrefix, entry.getKey()); try { entry.getValue().close(); } catch (final Exception e) { if (firstException == null) { firstException = new ProcessorStateException(String.format("%s Failed to close state store %s", logPrefix, entry.getKey()), e); } log.error("{} Failed to close state store {}: ", logPrefix, entry.getKey(), e); } } if (ackedOffsets != null) { checkpoint(ackedOffsets); } } } finally { try { stateDirectory.unlock(taskId); } catch (final IOException e) { if (firstException == null) { firstException = new ProcessorStateException(String.format("%s Failed to release state dir lock", logPrefix), e); } log.error("{} Failed to release state dir lock: ", logPrefix, e); } } if (firstException != null) { throw firstException; } } ProcessorStateManager(final TaskId taskId,
                                 final Collection<TopicPartition> sources,
                                 final boolean isStandby,
                                 final StateDirectory stateDirectory,
                                 final Map<String, String> storeToChangelogTopic,
                                 final ChangelogReader changelogReader,
                                 final boolean eosEnabled); static String storeChangelogTopic(final String applicationId, final String storeName); @Override File baseDir(); @Override void register(final StateStore store,
                         final boolean loggingEnabled,
                         final StateRestoreCallback stateRestoreCallback); @Override Map<TopicPartition, Long> checkpointed(); @Override StateStore getStore(final String name); @Override void flush(); @Override void close(final Map<TopicPartition, Long> ackedOffsets); @Override void checkpoint(final Map<TopicPartition, Long> ackedOffsets); @Override StateStore getGlobalStore(final String name); static final String STATE_CHANGELOG_TOPIC_SUFFIX; }### Answer:
@Test public void shouldThrowLockExceptionIfFailedToLockStateDirectory() throws Exception { final File taskDirectory = stateDirectory.directoryForTask(taskId); final FileChannel channel = FileChannel.open(new File(taskDirectory, StateDirectory.LOCK_FILE_NAME).toPath(), StandardOpenOption.CREATE, StandardOpenOption.WRITE); final FileLock lock = channel.lock(); try { new ProcessorStateManager( taskId, noPartitions, false, stateDirectory, Collections.<String, String>emptyMap(), changelogReader, false); fail("Should have thrown LockException"); } catch (final LockException e) { } finally { lock.release(); channel.close(); } }
@Test public void shouldDeleteCheckpointFileOnCreationIfEosEnabled() throws Exception { checkpoint.write(Collections.<TopicPartition, Long>emptyMap()); assertTrue(checkpointFile.exists()); ProcessorStateManager stateManager = null; try { stateManager = new ProcessorStateManager( taskId, noPartitions, false, stateDirectory, Collections.<String, String>emptyMap(), changelogReader, true); assertFalse(checkpointFile.exists()); } finally { if (stateManager != null) { stateManager.close(null); } } }