StorageConfig extends MapConfig { public List<String> getStoreNames() { Config subConfig = subset(STORE_PREFIX, true); List<String> storeNames = new ArrayList<>(); for (String key : subConfig.keySet()) { if (key.endsWith(SIDE_INPUT_PROCESSOR_FACTORY_SUFFIX)) { storeNames.add(key.substring(0, key.length() - SIDE_INPUT_PROCESSOR_FACTORY_SUFFIX.length())); } else if (key.endsWith(FACTORY_SUFFIX)) { storeNames.add(key.substring(0, key.length() - FACTORY_SUFFIX.length())); } } return storeNames; } StorageConfig(Config config); List<String> getStoreNames(); Optional<String> getChangelogStream(String storeName); boolean getAccessLogEnabled(String storeName); String getAccessLogStream(String changeLogStream); int getAccessLogSamplingRatio(String storeName); Optional<String> getStorageFactoryClassName(String storeName); Optional<String> getStorageKeySerde(String storeName); Optional<String> getStorageMsgSerde(String storeName); List<String> getSideInputs(String storeName); Optional<String> getSideInputsProcessorFactory(String storeName); Optional<String> getSideInputsProcessorSerializedInstance(String storeName); long getChangeLogDeleteRetentionInMs(String storeName); int getChangelogMaxMsgSizeBytes(String storeName); boolean getDisallowLargeMessages(String storeName); boolean getDropLargeMessages(String storeName); long getChangelogMinCompactionLagMs(String storeName); boolean isChangelogSystem(String systemName); boolean hasDurableStores(); int getNumPersistentStores(); boolean getCleanLoggedStoreDirsOnStart(String storeName); static final String FACTORY; static final String KEY_SERDE; static final String MSG_SERDE; static final String CHANGELOG_STREAM; static final String ACCESSLOG_STREAM_SUFFIX; static final String CHANGELOG_REPLICATION_FACTOR; static final String CHANGELOG_MAX_MSG_SIZE_BYTES; static final int DEFAULT_CHANGELOG_MAX_MSG_SIZE_BYTES; static final String DISALLOW_LARGE_MESSAGES; static final boolean DEFAULT_DISALLOW_LARGE_MESSAGES; static final String DROP_LARGE_MESSAGES; static final boolean DEFAULT_DROP_LARGE_MESSAGES; static final String MIN_COMPACTION_LAG_MS; static final String CHANGELOG_MIN_COMPACTION_LAG_MS; static final long DEFAULT_CHANGELOG_MIN_COMPACTION_LAG_MS; }
@Test public void testGetStoreNames() { assertEquals(Collections.emptyList(), new StorageConfig(new MapConfig()).getStoreNames()); Set<String> expectedStoreNames = ImmutableSet.of(STORE_NAME0, STORE_NAME1); StorageConfig storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(StorageConfig.FACTORY, STORE_NAME0), "store0.factory.class", String.format(StorageConfig.FACTORY, STORE_NAME1), "store1.factory.class"))); List<String> actual = storageConfig.getStoreNames(); assertEquals(2, actual.size()); assertEquals(expectedStoreNames, ImmutableSet.copyOf(actual)); StorageConfig config = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(FACTORY, STORE_NAME0), "store0.factory.class", String.format(StorageConfig.SIDE_INPUTS_PROCESSOR_FACTORY, STORE_NAME1), "store1.factory.class"))); actual = storageConfig.getStoreNames(); assertEquals(2, actual.size()); assertEquals(expectedStoreNames, ImmutableSet.copyOf(actual)); }