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 testClose() throws IOException {
          File baseDir = Files.createTempDirectory(stateDir).toFile();
          File checkpointFile = new File(baseDir, ProcessorStateManager.CHECKPOINT_FILE_NAME);
          try {
              // write an empty checkpoint file
              OffsetCheckpoint oldCheckpoint = new OffsetCheckpoint(checkpointFile);
              oldCheckpoint.write(Collections.<TopicPartition, Long>emptyMap());
  
              MockRestoreConsumer restoreConsumer = new MockRestoreConsumer();
  
              restoreConsumer.updatePartitions(persistentStoreTopicName, Utils.mkList(
                      new PartitionInfo(persistentStoreTopicName, 1, Node.noNode(), new Node[0], new Node[0])
              ));
              restoreConsumer.updatePartitions(nonPersistentStoreTopicName, Utils.mkList(
                      new PartitionInfo(nonPersistentStoreTopicName, 1, Node.noNode(), new Node[0], new Node[0])
              ));
  
              // set up ack'ed offsets
              HashMap<TopicPartition, Long> ackedOffsets = new HashMap<>();
              ackedOffsets.put(new TopicPartition(persistentStoreTopicName, 1), 123L);
              ackedOffsets.put(new TopicPartition(nonPersistentStoreTopicName, 1), 456L);
              ackedOffsets.put(new TopicPartition(ProcessorStateManager.storeChangelogTopic(applicationId, "otherTopic"), 1), 789L);
  
              MockStateStoreSupplier.MockStateStore persistentStore = new MockStateStoreSupplier.MockStateStore(persistentStoreName, true);
              MockStateStoreSupplier.MockStateStore nonPersistentStore = new MockStateStoreSupplier.MockStateStore(nonPersistentStoreName, false);
  
              ProcessorStateManager stateMgr = new ProcessorStateManager(applicationId, 1, noPartitions, baseDir, restoreConsumer, false);
              try {
                  // make sure the checkpoint file is deleted
                  assertFalse(checkpointFile.exists());
  
                  restoreConsumer.reset();
                  stateMgr.register(persistentStore, true, persistentStore.stateRestoreCallback);
  
                  restoreConsumer.reset();
                  stateMgr.register(nonPersistentStore, true, nonPersistentStore.stateRestoreCallback);
              } finally {
                  // close the state manager with the ack'ed offsets
                  stateMgr.close(ackedOffsets);
              }
  
              // make sure all stores are closed, and the checkpoint file is written.
              assertTrue(persistentStore.flushed);
              assertTrue(persistentStore.closed);
              assertTrue(nonPersistentStore.flushed);
              assertTrue(nonPersistentStore.closed);
              assertTrue(checkpointFile.exists());
  
              // the checkpoint file should contain an offset from the persistent store only.
              OffsetCheckpoint newCheckpoint = new OffsetCheckpoint(checkpointFile);
              Map<TopicPartition, Long> checkpointedOffsets = newCheckpoint.read();
              assertEquals(1, checkpointedOffsets.size());
              assertEquals(new Long(123L + 1L), checkpointedOffsets.get(new TopicPartition(persistentStoreTopicName, 1)));
          } finally {
              Utils.delete(baseDir);
          }
      }
test_case.class_method_signature: ProcessorStateManagerTest.testClose()
test_case.full_signature: @Test public void testClose()
test_case.identifier: testClose
test_case.invocations: toFile createTempDirectory write emptyMap updatePartitions mkList noNode updatePartitions mkList noNode put put put storeChangelogTopic assertFalse exists reset register reset register close assertTrue assertTrue assertTrue assertTrue assertTrue exists read assertEquals size assertEquals get delete
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testClose()
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 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();
          }
      }
focal_method.class_method_signature: ProcessorStateManager.close(Map<TopicPartition, Long> ackedOffsets)
focal_method.full_signature: public void close(Map<TopicPartition, Long> ackedOffsets)
focal_method.identifier: close
focal_method.invocations: isEmpty debug entrySet debug getKey flush getValue close getValue keySet contains storeChangelogTopic getPartition getPartition persistent get get put get put write release close channel
focal_method.modifiers: public
focal_method.parameters: (Map<TopicPartition, Long> ackedOffsets)
focal_method.return: void
focal_method.signature: void close(Map<TopicPartition, Long> ackedOffsets)