### Question:
GlobalStateManagerImpl implements GlobalStateManager { @Override public Set<String> initialize(final InternalProcessorContext processorContext) { try { if (!stateDirectory.lockGlobalState(MAX_LOCK_ATTEMPTS)) { throw new LockException(String.format("Failed to lock the global state directory: %s", baseDir)); } } catch (IOException e) { throw new LockException(String.format("Failed to lock the global state directory: %s", baseDir)); } try { this.checkpointableOffsets.putAll(checkpoint.read()); } catch (IOException e) { try { stateDirectory.unlockGlobalState(); } catch (IOException e1) { log.error("failed to unlock the global state directory", e); } throw new StreamsException("Failed to read checkpoints for global state stores", e); } final List<StateStore> stateStores = topology.globalStateStores(); for (final StateStore stateStore : stateStores) { globalStoreNames.add(stateStore.name()); stateStore.init(processorContext, stateStore); } return Collections.unmodifiableSet(globalStoreNames); } GlobalStateManagerImpl(final ProcessorTopology topology,
                                  final Consumer<byte[], byte[]> consumer,
                                  final StateDirectory stateDirectory); @Override Set<String> initialize(final InternalProcessorContext processorContext); @Override StateStore getGlobalStore(final String name); @Override StateStore getStore(final String name); File baseDir(); void register(final StateStore store,
                         final boolean ignored,
                         final StateRestoreCallback stateRestoreCallback); @Override void flush(); @Override void close(final Map<TopicPartition, Long> offsets); @Override void checkpoint(final Map<TopicPartition, Long> offsets); @Override Map<TopicPartition, Long> checkpointed();  }### Answer:
@Test public void shouldLockGlobalStateDirectory() throws Exception { stateManager.initialize(context); assertTrue(new File(stateDirectory.globalStateDir(), ".lock").exists()); }
@Test(expected = LockException.class) public void shouldThrowLockExceptionIfCantGetLock() throws Exception { final StateDirectory stateDir = new StateDirectory("appId", stateDirPath, time); try { stateDir.lockGlobalState(1); stateManager.initialize(context); } finally { stateDir.unlockGlobalState(); } }
@Test public void shouldNotDeleteCheckpointFileAfterLoaded() throws Exception { writeCheckpoint(); stateManager.initialize(context); assertTrue(checkpointFile.exists()); }
@Test(expected = StreamsException.class) public void shouldThrowStreamsExceptionIfFailedToReadCheckpointedOffsets() throws Exception { writeCorruptCheckpoint(); stateManager.initialize(context); }
@Test public void shouldInitializeStateStores() throws Exception { stateManager.initialize(context); assertTrue(store1.initialized); assertTrue(store2.initialized); }
@Test public void shouldReturnInitializedStoreNames() throws Exception { final Set<String> storeNames = stateManager.initialize(context); assertEquals(Utils.mkSet(store1.name(), store2.name()), storeNames); }
@Test public void shouldReleaseLockIfExceptionWhenLoadingCheckpoints() throws Exception { writeCorruptCheckpoint(); try { stateManager.initialize(context); } catch (StreamsException e) { } final StateDirectory stateDir = new StateDirectory("appId", stateDirPath, new MockTime()); try { assertTrue(stateDir.lockGlobalState(1)); } finally { stateDir.unlockGlobalState(); } }
@Test public void shouldThrowLockExceptionIfIOExceptionCaughtWhenTryingToLockStateDir() throws Exception { stateManager = new GlobalStateManagerImpl(topology, consumer, new StateDirectory("appId", stateDirPath, time) { @Override public boolean lockGlobalState(final int retry) throws IOException { throw new IOException("KABOOM!"); } }); try { stateManager.initialize(context); fail("Should have thrown LockException"); } catch (final LockException e) { } }