@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(); } }
@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 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 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); }
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(); }
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();  }