StorageConfig extends MapConfig { public Optional<String> getChangelogStream(String storeName) { String systemStream = StringUtils.trimToNull(get(String.format(CHANGELOG_STREAM, storeName), null)); String systemStreamRes; if (systemStream != null && !systemStream.contains(".")) { Optional<String> changelogSystem = getChangelogSystem(); if (changelogSystem.isPresent()) { systemStreamRes = changelogSystem.get() + "." + systemStream; } else { throw new SamzaException("changelog system is not defined:" + systemStream); } } else { systemStreamRes = systemStream; } if (systemStreamRes != null) { systemStreamRes = StreamManager.createUniqueNameForBatch(systemStreamRes, this); } return Optional.ofNullable(systemStreamRes); } 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 testGetChangelogStream() { assertEquals(Optional.empty(), new StorageConfig(new MapConfig()).getChangelogStream(STORE_NAME0)); StorageConfig storageConfig = new StorageConfig( new MapConfig(ImmutableMap.of(String.format(StorageConfig.CHANGELOG_STREAM, STORE_NAME0), ""))); assertEquals(Optional.empty(), storageConfig.getChangelogStream(STORE_NAME0)); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(StorageConfig.CHANGELOG_STREAM, STORE_NAME0), "changelog-system.changelog-stream0"))); assertEquals(Optional.of("changelog-system.changelog-stream0"), storageConfig.getChangelogStream(STORE_NAME0)); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(StorageConfig.CHANGELOG_STREAM, STORE_NAME0), "changelog-stream0", StorageConfig.CHANGELOG_SYSTEM, "changelog-system"))); assertEquals(Optional.of("changelog-system.changelog-stream0"), storageConfig.getChangelogStream(STORE_NAME0)); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(StorageConfig.CHANGELOG_STREAM, STORE_NAME0), "changelog-system.changelog-stream0", ApplicationConfig.APP_MODE, ApplicationConfig.ApplicationMode.BATCH.name().toLowerCase(), ApplicationConfig.APP_RUN_ID, "run-id"))); assertEquals(Optional.of("changelog-system.changelog-stream0-run-id"), storageConfig.getChangelogStream(STORE_NAME0)); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(StorageConfig.CHANGELOG_SYSTEM, "changelog-system", JobConfig.JOB_DEFAULT_SYSTEM, "should-not-be-used"))); assertEquals(Optional.empty(), storageConfig.getChangelogStream(STORE_NAME0)); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(StorageConfig.CHANGELOG_SYSTEM, "changelog-system", JobConfig.JOB_DEFAULT_SYSTEM, "should-not-be-used", String.format(CHANGELOG_STREAM, STORE_NAME0), "streamName"))); assertEquals("changelog-system.streamName", storageConfig.getChangelogStream(STORE_NAME0).get()); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(StorageConfig.CHANGELOG_SYSTEM, "changelog-system", JobConfig.JOB_DEFAULT_SYSTEM, "should-not-be-used", String.format(CHANGELOG_STREAM, STORE_NAME0), "changelog-system.streamName"))); assertEquals("changelog-system.streamName", storageConfig.getChangelogStream(STORE_NAME0).get()); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(StorageConfig.CHANGELOG_SYSTEM, "default-changelog-system", JobConfig.JOB_DEFAULT_SYSTEM, "default-system", String.format(CHANGELOG_STREAM, STORE_NAME0), "nondefault-changelog-system.streamName"))); assertEquals("nondefault-changelog-system.streamName", storageConfig.getChangelogStream(STORE_NAME0).get()); storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(JobConfig.JOB_DEFAULT_SYSTEM, "default-system", String.format(CHANGELOG_STREAM, STORE_NAME0), "streamName"))); assertEquals("default-system.streamName", storageConfig.getChangelogStream(STORE_NAME0).get()); }
@Test(expected = SamzaException.class) public void testGetChangelogStreamMissingSystem() { StorageConfig storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(StorageConfig.CHANGELOG_STREAM, STORE_NAME0), "changelog-stream0"))); storageConfig.getChangelogStream(STORE_NAME0); }