StorageConfig extends MapConfig { public boolean isChangelogSystem(String systemName) { return getStoreNames().stream() .map(this::getChangelogStream) .filter(Optional::isPresent) .map(systemStreamName -> StreamUtil.getSystemStreamFromNames(systemStreamName.get()).getSystem()) .anyMatch(system -> system.equals(systemName)); } 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 testIsChangelogSystem() { StorageConfig storageConfig = new StorageConfig(new MapConfig(ImmutableMap.of( String.format(StorageConfig.FACTORY, STORE_NAME0), "factory.class", String.format(CHANGELOG_STREAM, STORE_NAME0), "system0.changelog-stream", String.format(StorageConfig.FACTORY, STORE_NAME1), "factory.class"))); assertTrue(storageConfig.isChangelogSystem("system0")); assertFalse(storageConfig.isChangelogSystem("other-system")); }