ProcessorStateManager { public static FileLock lockStateDirectory(File stateDir) throws IOException { return lockStateDirectory(stateDir, 0); } ProcessorStateManager(String applicationId, int defaultPartition, Collection<TopicPartition> sources, File baseDir, Consumer<byte[], byte[]> restoreConsumer, boolean isStandby); static String storeChangelogTopic(String applicationId, String storeName); static FileLock lockStateDirectory(File stateDir); File baseDir(); void register(StateStore store, boolean loggingEnabled, StateRestoreCallback stateRestoreCallback); Map<TopicPartition, Long> checkpointedOffsets(); List<ConsumerRecord<byte[], byte[]>> updateStandbyStates(TopicPartition storePartition, List<ConsumerRecord<byte[], byte[]>> records); void putOffsetLimit(TopicPartition partition, long limit); StateStore getStore(String name); void flush(); void close(Map<TopicPartition, Long> ackedOffsets); static final String STATE_CHANGELOG_TOPIC_SUFFIX; static final String CHECKPOINT_FILE_NAME; static final String LOCK_FILE_NAME; }
@Test public void testLockStateDirectory() throws IOException { File baseDir = Files.createTempDirectory(stateDir).toFile(); try { FileLock lock; ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, 1, noPartitions, baseDir, new MockRestoreConsumer(), false); try { lock = ProcessorStateManager.lockStateDirectory(baseDir); assertNull(lock); } finally { stateMgr.close(Collections.<TopicPartition, Long>emptyMap()); } lock = ProcessorStateManager.lockStateDirectory(baseDir); try { assertNotNull(lock); } finally { if (lock != null) { lock.release(); lock.channel().close(); } } } finally { Utils.delete(baseDir); } }