@Test public void shouldThrowIllegalArgumentExceptionOnRegisterWhenStoreHasAlreadyBeenRegistered() throws Exception { final ProcessorStateManager stateManager = new ProcessorStateManager( taskId, noPartitions, false, stateDirectory, Collections.<String, String>emptyMap(), changelogReader, false); stateManager.register(mockStateStore, false, null); try { stateManager.register(mockStateStore, false, null); fail("should have thrown illegal argument exception when store with same name already registered"); } catch (final IllegalArgumentException e) { } }
@Override public void register(final StateStore store, final boolean loggingEnabled, final StateRestoreCallback stateRestoreCallback) { log.debug("{} Registering state store {} to its state manager", logPrefix, store.name()); if (store.name().equals(CHECKPOINT_FILE_NAME)) { throw new IllegalArgumentException(String.format("%s Illegal store name: %s", logPrefix, CHECKPOINT_FILE_NAME)); } if (stores.containsKey(store.name())) { throw new IllegalArgumentException(String.format("%s Store %s has already been registered.", logPrefix, store.name())); } final String topic = storeToChangelogTopic.get(store.name()); if (topic == null) { stores.put(store.name(), store); return; } final TopicPartition storePartition = new TopicPartition(topic, getPartition(topic)); changelogReader.validatePartitionExists(storePartition, store.name()); if (isStandby) { if (store.persistent()) { log.trace("{} Preparing standby replica of persistent state store {} with changelog topic {}", logPrefix, store.name(), topic); restoreCallbacks.put(topic, stateRestoreCallback); } } else { log.trace("{} Restoring state store {} from changelog topic {}", logPrefix, store.name(), topic); final StateRestorer restorer = new StateRestorer(storePartition, stateRestoreCallback, checkpointedOffsets.get(storePartition), offsetLimit(storePartition), store.persistent()); changelogReader.register(restorer); } stores.put(store.name(), store); }
ProcessorStateManager implements StateManager { @Override public void register(final StateStore store, final boolean loggingEnabled, final StateRestoreCallback stateRestoreCallback) { log.debug("{} Registering state store {} to its state manager", logPrefix, store.name()); if (store.name().equals(CHECKPOINT_FILE_NAME)) { throw new IllegalArgumentException(String.format("%s Illegal store name: %s", logPrefix, CHECKPOINT_FILE_NAME)); } if (stores.containsKey(store.name())) { throw new IllegalArgumentException(String.format("%s Store %s has already been registered.", logPrefix, store.name())); } final String topic = storeToChangelogTopic.get(store.name()); if (topic == null) { stores.put(store.name(), store); return; } final TopicPartition storePartition = new TopicPartition(topic, getPartition(topic)); changelogReader.validatePartitionExists(storePartition, store.name()); if (isStandby) { if (store.persistent()) { log.trace("{} Preparing standby replica of persistent state store {} with changelog topic {}", logPrefix, store.name(), topic); restoreCallbacks.put(topic, stateRestoreCallback); } } else { log.trace("{} Restoring state store {} from changelog topic {}", logPrefix, store.name(), topic); final StateRestorer restorer = new StateRestorer(storePartition, stateRestoreCallback, checkpointedOffsets.get(storePartition), offsetLimit(storePartition), store.persistent()); changelogReader.register(restorer); } stores.put(store.name(), store); } }
ProcessorStateManager implements StateManager { @Override public void register(final StateStore store, final boolean loggingEnabled, final StateRestoreCallback stateRestoreCallback) { log.debug("{} Registering state store {} to its state manager", logPrefix, store.name()); if (store.name().equals(CHECKPOINT_FILE_NAME)) { throw new IllegalArgumentException(String.format("%s Illegal store name: %s", logPrefix, CHECKPOINT_FILE_NAME)); } if (stores.containsKey(store.name())) { throw new IllegalArgumentException(String.format("%s Store %s has already been registered.", logPrefix, store.name())); } final String topic = storeToChangelogTopic.get(store.name()); if (topic == null) { stores.put(store.name(), store); return; } final TopicPartition storePartition = new TopicPartition(topic, getPartition(topic)); changelogReader.validatePartitionExists(storePartition, store.name()); if (isStandby) { if (store.persistent()) { log.trace("{} Preparing standby replica of persistent state store {} with changelog topic {}", logPrefix, store.name(), topic); restoreCallbacks.put(topic, stateRestoreCallback); } } else { log.trace("{} Restoring state store {} from changelog topic {}", logPrefix, store.name(), topic); final StateRestorer restorer = new StateRestorer(storePartition, stateRestoreCallback, checkpointedOffsets.get(storePartition), offsetLimit(storePartition), store.persistent()); changelogReader.register(restorer); } stores.put(store.name(), store); } ProcessorStateManager(final TaskId taskId,
                                 final Collection<TopicPartition> sources,
                                 final boolean isStandby,
                                 final StateDirectory stateDirectory,
                                 final Map<String, String> storeToChangelogTopic,
                                 final ChangelogReader changelogReader,
                                 final boolean eosEnabled); }
ProcessorStateManager implements StateManager { @Override public void register(final StateStore store, final boolean loggingEnabled, final StateRestoreCallback stateRestoreCallback) { log.debug("{} Registering state store {} to its state manager", logPrefix, store.name()); if (store.name().equals(CHECKPOINT_FILE_NAME)) { throw new IllegalArgumentException(String.format("%s Illegal store name: %s", logPrefix, CHECKPOINT_FILE_NAME)); } if (stores.containsKey(store.name())) { throw new IllegalArgumentException(String.format("%s Store %s has already been registered.", logPrefix, store.name())); } final String topic = storeToChangelogTopic.get(store.name()); if (topic == null) { stores.put(store.name(), store); return; } final TopicPartition storePartition = new TopicPartition(topic, getPartition(topic)); changelogReader.validatePartitionExists(storePartition, store.name()); if (isStandby) { if (store.persistent()) { log.trace("{} Preparing standby replica of persistent state store {} with changelog topic {}", logPrefix, store.name(), topic); restoreCallbacks.put(topic, stateRestoreCallback); } } else { log.trace("{} Restoring state store {} from changelog topic {}", logPrefix, store.name(), topic); final StateRestorer restorer = new StateRestorer(storePartition, stateRestoreCallback, checkpointedOffsets.get(storePartition), offsetLimit(storePartition), store.persistent()); changelogReader.register(restorer); } stores.put(store.name(), store); } ProcessorStateManager(final TaskId taskId,
                                 final Collection<TopicPartition> sources,
                                 final boolean isStandby,
                                 final StateDirectory stateDirectory,
                                 final Map<String, String> storeToChangelogTopic,
                                 final ChangelogReader changelogReader,
                                 final boolean eosEnabled); static String storeChangelogTopic(final String applicationId, final String storeName); @Override File baseDir(); @Override void register(final StateStore store,
                         final boolean loggingEnabled,
                         final StateRestoreCallback stateRestoreCallback); @Override Map<TopicPartition, Long> checkpointed(); @Override StateStore getStore(final String name); @Override void flush(); @Override void close(final Map<TopicPartition, Long> ackedOffsets); @Override void checkpoint(final Map<TopicPartition, Long> ackedOffsets); @Override StateStore getGlobalStore(final String name); }
ProcessorStateManager implements StateManager { @Override public void register(final StateStore store, final boolean loggingEnabled, final StateRestoreCallback stateRestoreCallback) { log.debug("{} Registering state store {} to its state manager", logPrefix, store.name()); if (store.name().equals(CHECKPOINT_FILE_NAME)) { throw new IllegalArgumentException(String.format("%s Illegal store name: %s", logPrefix, CHECKPOINT_FILE_NAME)); } if (stores.containsKey(store.name())) { throw new IllegalArgumentException(String.format("%s Store %s has already been registered.", logPrefix, store.name())); } final String topic = storeToChangelogTopic.get(store.name()); if (topic == null) { stores.put(store.name(), store); return; } final TopicPartition storePartition = new TopicPartition(topic, getPartition(topic)); changelogReader.validatePartitionExists(storePartition, store.name()); if (isStandby) { if (store.persistent()) { log.trace("{} Preparing standby replica of persistent state store {} with changelog topic {}", logPrefix, store.name(), topic); restoreCallbacks.put(topic, stateRestoreCallback); } } else { log.trace("{} Restoring state store {} from changelog topic {}", logPrefix, store.name(), topic); final StateRestorer restorer = new StateRestorer(storePartition, stateRestoreCallback, checkpointedOffsets.get(storePartition), offsetLimit(storePartition), store.persistent()); changelogReader.register(restorer); } stores.put(store.name(), store); } ProcessorStateManager(final TaskId taskId,
                                 final Collection<TopicPartition> sources,
                                 final boolean isStandby,
                                 final StateDirectory stateDirectory,
                                 final Map<String, String> storeToChangelogTopic,
                                 final ChangelogReader changelogReader,
                                 final boolean eosEnabled); static String storeChangelogTopic(final String applicationId, final String storeName); @Override File baseDir(); @Override void register(final StateStore store,
                         final boolean loggingEnabled,
                         final StateRestoreCallback stateRestoreCallback); @Override Map<TopicPartition, Long> checkpointed(); @Override StateStore getStore(final String name); @Override void flush(); @Override void close(final Map<TopicPartition, Long> ackedOffsets); @Override void checkpoint(final Map<TopicPartition, Long> ackedOffsets); @Override StateStore getGlobalStore(final String name); static final String STATE_CHANGELOG_TOPIC_SUFFIX; }