StorageConfig extends MapConfig { public long getChangelogMinCompactionLagMs(String storeName) { String minCompactLagConfigName = String.format(CHANGELOG_MIN_COMPACTION_LAG_MS, storeName); checkArgument(get("stores." + storeName + ".changelog.kafka." + MIN_COMPACTION_LAG_MS) == null, "Use " + minCompactLagConfigName + " to set kafka min.compaction.lag.ms property."); return getLong(minCompactLagConfigName, getDefaultChangelogMinCompactionLagMs()); } 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 testGetChangelogMinCompactionLagMs() { Map<String, String> configMap = new HashMap<>(); assertEquals(DEFAULT_CHANGELOG_MIN_COMPACTION_LAG_MS, new StorageConfig(new MapConfig(configMap)).getChangelogMinCompactionLagMs(STORE_NAME0)); long defaultLagOverride = TimeUnit.HOURS.toMillis(8); configMap.put(String.format(CHANGELOG_MIN_COMPACTION_LAG_MS, "default"), String.valueOf(defaultLagOverride)); assertEquals(defaultLagOverride, new StorageConfig(new MapConfig(configMap)).getChangelogMinCompactionLagMs(STORE_NAME0)); long storeSpecificLagOverride = TimeUnit.HOURS.toMillis(6); configMap.put(String.format(CHANGELOG_MIN_COMPACTION_LAG_MS, STORE_NAME0), String.valueOf(storeSpecificLagOverride)); assertEquals(storeSpecificLagOverride, new StorageConfig(new MapConfig(configMap)).getChangelogMinCompactionLagMs(STORE_NAME0)); }