StorageConfig extends MapConfig { public List<String> getSideInputs(String storeName) { return Optional.ofNullable(get(String.format(SIDE_INPUTS, storeName), null)) .map(inputs -> Stream.of(inputs.split(",")) .map(String::trim) .filter(input -> !input.isEmpty()) .collect(Collectors.toList())) .orElse(Collections.emptyList()); } 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 testGetSideInputs() { assertEquals(Collections.emptyList(), new StorageConfig(new MapConfig()).getSideInputs(STORE_NAME0)); StorageConfig storageConfig = new StorageConfig( new MapConfig(ImmutableMap.of(String.format(StorageConfig.SIDE_INPUTS, STORE_NAME0), "side-input"))); assertEquals(Collections.singletonList("side-input"), storageConfig.getSideInputs(STORE_NAME0)); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(StorageConfig.SIDE_INPUTS, STORE_NAME0), "side-input0,side-input1"))); assertEquals(ImmutableList.of("side-input0", "side-input1"), storageConfig.getSideInputs(STORE_NAME0)); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(StorageConfig.SIDE_INPUTS, STORE_NAME0), ", side-input0 ,,side-input1,"))); assertEquals(ImmutableList.of("side-input0", "side-input1"), storageConfig.getSideInputs(STORE_NAME0)); }