test_class.declarator: noPartitions = Collections.emptySet()
test_class.modifier: private final
test_class.original_string: private final Set<TopicPartition> noPartitions = Collections.emptySet();
test_class.type: Set<TopicPartition>
test_class.var_name: noPartitions
test_class.declarator: applicationId = "test-application"
test_class.modifier: private final
test_class.original_string: private final String applicationId = "test-application";
test_class.type: String
test_class.var_name: applicationId
test_class.declarator: stateDir = "test"
test_class.modifier: private final
test_class.original_string: private final String stateDir = "test";
test_class.type: String
test_class.var_name: stateDir
test_class.declarator: persistentStoreName = "persistentStore"
test_class.modifier: private final
test_class.original_string: private final String persistentStoreName = "persistentStore";
test_class.type: String
test_class.var_name: persistentStoreName
test_class.declarator: nonPersistentStoreName = "nonPersistentStore"
test_class.modifier: private final
test_class.original_string: private final String nonPersistentStoreName = "nonPersistentStore";
test_class.type: String
test_class.var_name: nonPersistentStoreName
test_class.declarator: persistentStoreTopicName = ProcessorStateManager.storeChangelogTopic(applicationId, persistentStoreName)
test_class.modifier: private final
test_class.original_string: private final String persistentStoreTopicName = ProcessorStateManager.storeChangelogTopic(applicationId, persistentStoreName);
test_class.type: String
test_class.var_name: persistentStoreTopicName
test_class.declarator: nonPersistentStoreTopicName = ProcessorStateManager.storeChangelogTopic(applicationId, nonPersistentStoreName)
test_class.modifier: private final
test_class.original_string: private final String nonPersistentStoreTopicName = ProcessorStateManager.storeChangelogTopic(applicationId, nonPersistentStoreName);
test_class.type: String
test_class.var_name: nonPersistentStoreTopicName
test_class.file: streams/src/test/java/org/apache/kafka/streams/processor/internals/ProcessorStateManagerTest.java
test_class.identifier: ProcessorStateManagerTest
test_case.body: @Test
      public void testLockStateDirectory() throws IOException {
          File baseDir = Files.createTempDirectory(stateDir).toFile();
          try {
              FileLock lock;
  
              // the state manager locks the directory
              ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, 1, noPartitions, baseDir, new MockRestoreConsumer(), false);
  
              try {
                  // this should not get the lock
                  lock = ProcessorStateManager.lockStateDirectory(baseDir);
                  assertNull(lock);
              } finally {
                  // by closing the state manager, release the lock
                  stateMgr.close(Collections.<TopicPartition, Long>emptyMap());
              }
  
              // now, this should get the lock
              lock = ProcessorStateManager.lockStateDirectory(baseDir);
              try {
                  assertNotNull(lock);
              } finally {
                  if (lock != null) {
                      lock.release();
                      lock.channel().close();
                  }
              }
          } finally {
              Utils.delete(baseDir);
          }
      }
test_case.class_method_signature: ProcessorStateManagerTest.testLockStateDirectory()
test_case.full_signature: @Test public void testLockStateDirectory()
test_case.identifier: testLockStateDirectory
test_case.invocations: toFile createTempDirectory lockStateDirectory assertNull close emptyMap lockStateDirectory assertNotNull release close channel delete
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testLockStateDirectory()
focal_class.declarator: log = LoggerFactory.getLogger(ProcessorStateManager.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger log = LoggerFactory.getLogger(ProcessorStateManager.class);
focal_class.type: Logger
focal_class.var_name: log
focal_class.declarator: STATE_CHANGELOG_TOPIC_SUFFIX = "-changelog"
focal_class.modifier: public static final
focal_class.original_string: public static final String STATE_CHANGELOG_TOPIC_SUFFIX = "-changelog";
focal_class.type: String
focal_class.var_name: STATE_CHANGELOG_TOPIC_SUFFIX
focal_class.declarator: CHECKPOINT_FILE_NAME = ".checkpoint"
focal_class.modifier: public static final
focal_class.original_string: public static final String CHECKPOINT_FILE_NAME = ".checkpoint";
focal_class.type: String
focal_class.var_name: CHECKPOINT_FILE_NAME
focal_class.declarator: LOCK_FILE_NAME = ".lock"
focal_class.modifier: public static final
focal_class.original_string: public static final String LOCK_FILE_NAME = ".lock";
focal_class.type: String
focal_class.var_name: LOCK_FILE_NAME
focal_class.declarator: applicationId
focal_class.modifier: private final
focal_class.original_string: private final String applicationId;
focal_class.type: String
focal_class.var_name: applicationId
focal_class.declarator: defaultPartition
focal_class.modifier: private final
focal_class.original_string: private final int defaultPartition;
focal_class.type: int
focal_class.var_name: defaultPartition
focal_class.declarator: partitionForTopic
focal_class.modifier: private final
focal_class.original_string: private final Map<String, TopicPartition> partitionForTopic;
focal_class.type: Map<String, TopicPartition>
focal_class.var_name: partitionForTopic
focal_class.declarator: baseDir
focal_class.modifier: private final
focal_class.original_string: private final File baseDir;
focal_class.type: File
focal_class.var_name: baseDir
focal_class.declarator: directoryLock
focal_class.modifier: private final
focal_class.original_string: private final FileLock directoryLock;
focal_class.type: FileLock
focal_class.var_name: directoryLock
focal_class.declarator: stores
focal_class.modifier: private final
focal_class.original_string: private final Map<String, StateStore> stores;
focal_class.type: Map<String, StateStore>
focal_class.var_name: stores
focal_class.declarator: loggingEnabled
focal_class.modifier: private final
focal_class.original_string: private final Set<String> loggingEnabled;
focal_class.type: Set<String>
focal_class.var_name: loggingEnabled
focal_class.declarator: restoreConsumer
focal_class.modifier: private final
focal_class.original_string: private final Consumer<byte[], byte[]> restoreConsumer;
focal_class.type: Consumer<byte[], byte[]>
focal_class.var_name: restoreConsumer
focal_class.declarator: restoredOffsets
focal_class.modifier: private final
focal_class.original_string: private final Map<TopicPartition, Long> restoredOffsets;
focal_class.type: Map<TopicPartition, Long>
focal_class.var_name: restoredOffsets
focal_class.declarator: checkpointedOffsets
focal_class.modifier: private final
focal_class.original_string: private final Map<TopicPartition, Long> checkpointedOffsets;
focal_class.type: Map<TopicPartition, Long>
focal_class.var_name: checkpointedOffsets
focal_class.declarator: offsetLimits
focal_class.modifier: private final
focal_class.original_string: private final Map<TopicPartition, Long> offsetLimits;
focal_class.type: Map<TopicPartition, Long>
focal_class.var_name: offsetLimits
focal_class.declarator: isStandby
focal_class.modifier: private final
focal_class.original_string: private final boolean isStandby;
focal_class.type: boolean
focal_class.var_name: isStandby
focal_class.declarator: restoreCallbacks
focal_class.modifier: private final
focal_class.original_string: private final Map<String, StateRestoreCallback> restoreCallbacks;
focal_class.type: Map<String, StateRestoreCallback>
focal_class.var_name: restoreCallbacks
focal_class.file: streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java
focal_class.identifier: ProcessorStateManager
focal_class.class_method_signature: ProcessorStateManager.ProcessorStateManager(String applicationId, int defaultPartition, Collection<TopicPartition> sources, File baseDir, Consumer<byte[], byte[]> restoreConsumer, boolean isStandby)
focal_class.full_signature: public  ProcessorStateManager(String applicationId, int defaultPartition, Collection<TopicPartition> sources, File baseDir, Consumer<byte[], byte[]> restoreConsumer, boolean isStandby)
focal_class.identifier: ProcessorStateManager
focal_class.modifiers: public
focal_class.parameters: (String applicationId, int defaultPartition, Collection<TopicPartition> sources, File baseDir, Consumer<byte[], byte[]> restoreConsumer, boolean isStandby)
focal_class.signature:  ProcessorStateManager(String applicationId, int defaultPartition, Collection<TopicPartition> sources, File baseDir, Consumer<byte[], byte[]> restoreConsumer, boolean isStandby)
focal_class.class_method_signature: ProcessorStateManager.createStateDirectory(File stateDir)
focal_class.full_signature: private static void createStateDirectory(File stateDir)
focal_class.identifier: createStateDirectory
focal_class.modifiers: private static
focal_class.parameters: (File stateDir)
focal_class.return: void
focal_class.signature: void createStateDirectory(File stateDir)
focal_class.class_method_signature: ProcessorStateManager.storeChangelogTopic(String applicationId, String storeName)
focal_class.full_signature: public static String storeChangelogTopic(String applicationId, String storeName)
focal_class.identifier: storeChangelogTopic
focal_class.modifiers: public static
focal_class.parameters: (String applicationId, String storeName)
focal_class.return: String
focal_class.signature: String storeChangelogTopic(String applicationId, String storeName)
focal_class.class_method_signature: ProcessorStateManager.lockStateDirectory(File stateDir)
focal_class.full_signature: public static FileLock lockStateDirectory(File stateDir)
focal_class.identifier: lockStateDirectory
focal_class.modifiers: public static
focal_class.parameters: (File stateDir)
focal_class.return: FileLock
focal_class.signature: FileLock lockStateDirectory(File stateDir)
focal_class.class_method_signature: ProcessorStateManager.lockStateDirectory(File stateDir, int retry)
focal_class.full_signature: private static FileLock lockStateDirectory(File stateDir, int retry)
focal_class.identifier: lockStateDirectory
focal_class.modifiers: private static
focal_class.parameters: (File stateDir, int retry)
focal_class.return: FileLock
focal_class.signature: FileLock lockStateDirectory(File stateDir, int retry)
focal_class.class_method_signature: ProcessorStateManager.lockStateDirectory(FileChannel channel)
focal_class.full_signature: private static FileLock lockStateDirectory(FileChannel channel)
focal_class.identifier: lockStateDirectory
focal_class.modifiers: private static
focal_class.parameters: (FileChannel channel)
focal_class.return: FileLock
focal_class.signature: FileLock lockStateDirectory(FileChannel channel)
focal_class.class_method_signature: ProcessorStateManager.baseDir()
focal_class.full_signature: public File baseDir()
focal_class.identifier: baseDir
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: File
focal_class.signature: File baseDir()
focal_class.class_method_signature: ProcessorStateManager.register(StateStore store, boolean loggingEnabled, StateRestoreCallback stateRestoreCallback)
focal_class.full_signature: public void register(StateStore store, boolean loggingEnabled, StateRestoreCallback stateRestoreCallback)
focal_class.identifier: register
focal_class.modifiers: public
focal_class.parameters: (StateStore store, boolean loggingEnabled, StateRestoreCallback stateRestoreCallback)
focal_class.return: void
focal_class.signature: void register(StateStore store, boolean loggingEnabled, StateRestoreCallback stateRestoreCallback)
focal_class.class_method_signature: ProcessorStateManager.restoreActiveState(String topicName, StateRestoreCallback stateRestoreCallback)
focal_class.full_signature: private void restoreActiveState(String topicName, StateRestoreCallback stateRestoreCallback)
focal_class.identifier: restoreActiveState
focal_class.modifiers: private
focal_class.parameters: (String topicName, StateRestoreCallback stateRestoreCallback)
focal_class.return: void
focal_class.signature: void restoreActiveState(String topicName, StateRestoreCallback stateRestoreCallback)
focal_class.class_method_signature: ProcessorStateManager.checkpointedOffsets()
focal_class.full_signature: public Map<TopicPartition, Long> checkpointedOffsets()
focal_class.identifier: checkpointedOffsets
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Map<TopicPartition, Long>
focal_class.signature: Map<TopicPartition, Long> checkpointedOffsets()
focal_class.class_method_signature: ProcessorStateManager.updateStandbyStates(TopicPartition storePartition, List<ConsumerRecord<byte[], byte[]>> records)
focal_class.full_signature: public List<ConsumerRecord<byte[], byte[]>> updateStandbyStates(TopicPartition storePartition, List<ConsumerRecord<byte[], byte[]>> records)
focal_class.identifier: updateStandbyStates
focal_class.modifiers: public
focal_class.parameters: (TopicPartition storePartition, List<ConsumerRecord<byte[], byte[]>> records)
focal_class.return: List<ConsumerRecord<byte[], byte[]>>
focal_class.signature: List<ConsumerRecord<byte[], byte[]>> updateStandbyStates(TopicPartition storePartition, List<ConsumerRecord<byte[], byte[]>> records)
focal_class.class_method_signature: ProcessorStateManager.putOffsetLimit(TopicPartition partition, long limit)
focal_class.full_signature: public void putOffsetLimit(TopicPartition partition, long limit)
focal_class.identifier: putOffsetLimit
focal_class.modifiers: public
focal_class.parameters: (TopicPartition partition, long limit)
focal_class.return: void
focal_class.signature: void putOffsetLimit(TopicPartition partition, long limit)
focal_class.class_method_signature: ProcessorStateManager.offsetLimit(TopicPartition partition)
focal_class.full_signature: private long offsetLimit(TopicPartition partition)
focal_class.identifier: offsetLimit
focal_class.modifiers: private
focal_class.parameters: (TopicPartition partition)
focal_class.return: long
focal_class.signature: long offsetLimit(TopicPartition partition)
focal_class.class_method_signature: ProcessorStateManager.getStore(String name)
focal_class.full_signature: public StateStore getStore(String name)
focal_class.identifier: getStore
focal_class.modifiers: public
focal_class.parameters: (String name)
focal_class.return: StateStore
focal_class.signature: StateStore getStore(String name)
focal_class.class_method_signature: ProcessorStateManager.flush()
focal_class.full_signature: public void flush()
focal_class.identifier: flush
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void flush()
focal_class.class_method_signature: ProcessorStateManager.close(Map<TopicPartition, Long> ackedOffsets)
focal_class.full_signature: public void close(Map<TopicPartition, Long> ackedOffsets)
focal_class.identifier: close
focal_class.modifiers: public
focal_class.parameters: (Map<TopicPartition, Long> ackedOffsets)
focal_class.return: void
focal_class.signature: void close(Map<TopicPartition, Long> ackedOffsets)
focal_class.class_method_signature: ProcessorStateManager.getPartition(String topic)
focal_class.full_signature: private int getPartition(String topic)
focal_class.identifier: getPartition
focal_class.modifiers: private
focal_class.parameters: (String topic)
focal_class.return: int
focal_class.signature: int getPartition(String topic)
focal_method.body: public static FileLock lockStateDirectory(File stateDir) throws IOException {
          return lockStateDirectory(stateDir, 0);
      }
focal_method.class_method_signature: ProcessorStateManager.lockStateDirectory(File stateDir)
focal_method.full_signature: public static FileLock lockStateDirectory(File stateDir)
focal_method.identifier: lockStateDirectory
focal_method.invocations: lockStateDirectory
focal_method.modifiers: public static
focal_method.parameters: (File stateDir)
focal_method.return: FileLock
focal_method.signature: FileLock lockStateDirectory(File stateDir)