diff --git a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java index 7bc6cb4c72..86c7e7de3b 100644 --- a/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java +++ b/samza-core/src/main/java/org/apache/samza/config/StorageConfig.java @@ -148,7 +148,7 @@ public Optional getStorageMsgSerde(String storeName) { * * @return the name of the system to use by default for all changelogs, if defined. */ - public Optional getChangelogSystem() { + private Optional getChangelogSystem() { return Optional.ofNullable(get(CHANGELOG_SYSTEM, get(JobConfig.JOB_DEFAULT_SYSTEM))); } diff --git a/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java b/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java index 2cde5df560..e094de2333 100644 --- a/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java +++ b/samza-core/src/test/java/org/apache/samza/config/TestStorageConfig.java @@ -28,6 +28,7 @@ import org.apache.samza.SamzaException; import org.junit.Test; +import static org.apache.samza.config.StorageConfig.*; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; @@ -81,6 +82,37 @@ public void testGetChangelogStream() { ApplicationConfig.ApplicationMode.BATCH.name().toLowerCase(), ApplicationConfig.APP_RUN_ID, "run-id"))); assertEquals(Optional.of("changelog-system.changelog-stream0-run-id"), storageConfig.getChangelogStream(STORE_NAME0)); + + // job has no changelog stream defined + 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)); + + // job.changelog.system takes precedence over job.default.system when changelog is specified as just streamName + 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()); + + // job.changelog.system takes precedence over job.default.system when changelog is specified as {systemName}.{streamName} + 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()); + + // systemName specified using stores.{storeName}.changelog = {systemName}.{streamName} should take precedence even + // when job.changelog.system and job.default.system are specified + 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()); + + // fall back to job.default.system if job.changelog.system is not specified + 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) @@ -152,23 +184,6 @@ public void testGetStorageMsgSerde() { assertEquals(Optional.of("my.msg.serde.class"), storageConfig.getStorageMsgSerde(STORE_NAME0)); } - @Test - public void testGetChangelogSystem() { - // empty config, so no system - assertEquals(Optional.empty(), new StorageConfig(new MapConfig()).getChangelogSystem()); - - // job.changelog.system takes precedence over job.default.system - StorageConfig storageConfig = new StorageConfig(new MapConfig( - ImmutableMap.of(StorageConfig.CHANGELOG_SYSTEM, "changelog-system", JobConfig.JOB_DEFAULT_SYSTEM, - "should-not-be-used"))); - assertEquals(Optional.of("changelog-system"), storageConfig.getChangelogSystem()); - - // fall back to job.default.system if job.changelog.system is not specified - storageConfig = - new StorageConfig(new MapConfig(ImmutableMap.of(JobConfig.JOB_DEFAULT_SYSTEM, "default-system"))); - assertEquals(Optional.of("default-system"), storageConfig.getChangelogSystem()); - } - @Test public void testGetSideInputs() { // empty config, so no system @@ -232,7 +247,7 @@ public void testIsChangelogSystem() { StorageConfig storageConfig = new StorageConfig(new MapConfig(ImmutableMap.of( // store0 has a changelog stream String.format(StorageConfig.FACTORY, STORE_NAME0), "factory.class", - String.format(StorageConfig.CHANGELOG_STREAM, STORE_NAME0), "system0.changelog-stream", + String.format(CHANGELOG_STREAM, STORE_NAME0), "system0.changelog-stream", // store1 does not have a changelog stream String.format(StorageConfig.FACTORY, STORE_NAME1), "factory.class"))); assertTrue(storageConfig.isChangelogSystem("system0")); @@ -248,7 +263,7 @@ public void testHasDurableStores() { storageConfig = new StorageConfig(new MapConfig( ImmutableMap.of(String.format(StorageConfig.FACTORY, STORE_NAME0), "factory.class", - String.format(StorageConfig.CHANGELOG_STREAM, STORE_NAME0), "system0.changelog-stream"))); + String.format(CHANGELOG_STREAM, STORE_NAME0), "system0.changelog-stream"))); assertTrue(storageConfig.hasDurableStores()); } diff --git a/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java b/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java index c18c82dc2a..1986bea54f 100644 --- a/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java +++ b/samza-kafka/src/main/java/org/apache/samza/system/kafka/KafkaSystemAdmin.java @@ -546,6 +546,14 @@ public KafkaStreamSpec toKafkaSpec(StreamSpec spec) { kafkaSpec = kafkaSpec.copyWithProperties(properties); } else { kafkaSpec = KafkaStreamSpec.fromSpec(spec); + + // we check if there is a system-level rf config specified, else we use KafkaConfig.topic-default-rf + int replicationFactorFromSystemConfig = Integer.valueOf( + new KafkaConfig(config).getSystemDefaultReplicationFactor(spec.getSystemName(), + KafkaConfig.TOPIC_DEFAULT_REPLICATION_FACTOR())); + LOG.info("Using replication-factor: {} for StreamSpec: {}", replicationFactorFromSystemConfig, spec); + return new KafkaStreamSpec(kafkaSpec.getId(), kafkaSpec.getPhysicalName(), kafkaSpec.getSystemName(), + kafkaSpec.getPartitionCount(), replicationFactorFromSystemConfig, kafkaSpec.getProperties()); } return kafkaSpec; } diff --git a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala index b9543fbaab..993f0e4225 100644 --- a/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala +++ b/samza-kafka/src/main/scala/org/apache/samza/config/KafkaConfig.scala @@ -118,7 +118,7 @@ class KafkaConfig(config: Config) extends ScalaMapConfig(config) { Option(replicationFactor) } - private def getSystemDefaultReplicationFactor(systemName: String, defaultValue: String) = { + def getSystemDefaultReplicationFactor(systemName: String, defaultValue: String) = { val defaultReplicationFactor = new SystemConfig(config).getDefaultStreamProperties(systemName).getOrDefault(KafkaConfig.TOPIC_REPLICATION_FACTOR, defaultValue) defaultReplicationFactor } @@ -243,19 +243,35 @@ class KafkaConfig(config: Config) extends ScalaMapConfig(config) { /** * Gets the replication factor for the changelog topics. Uses the following precedence. * - * 1. If stores.myStore.changelog.replication.factor is configured, that value is used. - * 2. If systems.changelog-system.default.stream.replication.factor is configured, that value is used. - * 3. 2 - * - * Note that the changelog-system has a similar precedence. See [[StorageConfig]] + * 1. If stores.{storeName}.changelog.replication.factor is configured, that value is used. + * 2. If it is not configured, the value configured for stores.default.changelog.replication.factor is used. + * 3. If it is not configured, the RF value configured for the store's changelog's system, configured using + * stores.{storeName}.changelog={systemName}.{streamName}, is used. + * 4. If it is not configured, the value for the RF of job.changelog.system is used. + * 5. If it is not configured, the value for the RF of job.default.system is used. + * 6. If it is not configured, the RF is chosen as 2. */ - def getChangelogStreamReplicationFactor(name: String) = getOption(KafkaConfig.CHANGELOG_STREAM_REPLICATION_FACTOR format name).getOrElse(getDefaultChangelogStreamReplicationFactor) + def getChangelogStreamReplicationFactor(storeName: String) = { + var changelogRF = getOption(KafkaConfig.CHANGELOG_STREAM_REPLICATION_FACTOR format storeName) + + if(!changelogRF.isDefined) { + changelogRF = getOption(KafkaConfig.DEFAULT_CHANGELOG_STREAM_REPLICATION_FACTOR) + } - def getDefaultChangelogStreamReplicationFactor() = { - val changelogSystem = new StorageConfig(config).getChangelogSystem.orElse(null) - getOption(KafkaConfig.DEFAULT_CHANGELOG_STREAM_REPLICATION_FACTOR).getOrElse(getSystemDefaultReplicationFactor(changelogSystem, "2")) + if(!changelogRF.isDefined) { + val changelogSystemStream = new StorageConfig(config).getChangelogStream(storeName) + if (!changelogSystemStream.isPresent) { + throw new SamzaException("Cannot deduce replication factor. Changelog system-stream not defined for store " + storeName) + } + + val changelogSystem = StreamUtil.getSystemStreamFromNames(changelogSystemStream.get()).getSystem + changelogRF = Option.apply(getSystemDefaultReplicationFactor(changelogSystem, null)) + } + + changelogRF.getOrElse(KafkaConfig.TOPIC_DEFAULT_REPLICATION_FACTOR) } + /** * Gets the max message bytes for the changelog topics. Uses the following precedence. * @@ -268,7 +284,7 @@ class KafkaConfig(config: Config) extends ScalaMapConfig(config) { def getChangelogStreamMaxMessageByte(name: String) = getOption(KafkaConfig.CHANGELOG_MAX_MESSAGE_BYTES format name) match { case Some(maxMessageBytes) => maxMessageBytes case _ => - val changelogSystem = new StorageConfig(config).getChangelogSystem.orElse(null) + val changelogSystem = StreamUtil.getSystemStreamFromNames(JavaOptionals.toRichOptional(new StorageConfig(config).getChangelogStream(name)).toOption.getOrElse(throw new SamzaException("System-stream not defined for store:"+name))).getSystem val systemMaxMessageBytes = new SystemConfig(config).getDefaultStreamProperties(changelogSystem).getOrDefault(KafkaConfig.MAX_MESSAGE_BYTES, KafkaConfig.DEFAULT_LOG_COMPACT_TOPIC_MAX_MESSAGE_BYTES) systemMaxMessageBytes } diff --git a/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaConfig.scala b/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaConfig.scala index bb1b337326..d94f414b32 100644 --- a/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaConfig.scala +++ b/samza-kafka/src/test/scala/org/apache/samza/config/TestKafkaConfig.scala @@ -211,16 +211,26 @@ class TestKafkaConfig { kafkaProducerConfig.getProducerProperties } + @Test + def testGetSystemDefaultReplicationFactor(): Unit = { + assertEquals(KafkaConfig.TOPIC_DEFAULT_REPLICATION_FACTOR, new KafkaConfig(new MapConfig()).getSystemDefaultReplicationFactor("kafka-system",KafkaConfig.TOPIC_DEFAULT_REPLICATION_FACTOR)) + + props.setProperty("systems.kafka-system.default.stream.replication.factor", "8") + val mapConfig = new MapConfig(props.asScala.asJava) + val kafkaConfig = new KafkaConfig(mapConfig) + assertEquals("8", kafkaConfig.getSystemDefaultReplicationFactor("kafka-system","2")) + } + @Test def testChangeLogReplicationFactor() { props.setProperty("stores.store-with-override.changelog", "kafka-system.changelog-topic") props.setProperty("stores.store-with-override.changelog.replication.factor", "3") + props.setProperty("stores.default.changelog.replication.factor", "2") val mapConfig = new MapConfig(props.asScala.asJava) val kafkaConfig = new KafkaConfig(mapConfig) assertEquals("3", kafkaConfig.getChangelogStreamReplicationFactor("store-with-override")) assertEquals("2", kafkaConfig.getChangelogStreamReplicationFactor("store-without-override")) - assertEquals("2", kafkaConfig.getDefaultChangelogStreamReplicationFactor) } @Test @@ -231,11 +241,11 @@ class TestKafkaConfig { // Override the "default" default value props.setProperty("stores.default.changelog.replication.factor", "5") + val mapConfig = new MapConfig(props.asScala.asJava) val kafkaConfig = new KafkaConfig(mapConfig) assertEquals("4", kafkaConfig.getChangelogStreamReplicationFactor("store-with-override")) assertEquals("5", kafkaConfig.getChangelogStreamReplicationFactor("store-without-override")) - assertEquals("5", kafkaConfig.getDefaultChangelogStreamReplicationFactor) } @Test @@ -243,12 +253,12 @@ class TestKafkaConfig { props.setProperty(StorageConfig.CHANGELOG_SYSTEM, "kafka-system") props.setProperty("systems.kafka-system.default.stream.replication.factor", "8") props.setProperty("stores.store-with-override.changelog.replication.factor", "4") + props.setProperty("stores.store-without-override.changelog", "change-for-store-without-override") val mapConfig = new MapConfig(props.asScala.asJava) val kafkaConfig = new KafkaConfig(mapConfig) assertEquals("4", kafkaConfig.getChangelogStreamReplicationFactor("store-with-override")) assertEquals("8", kafkaConfig.getChangelogStreamReplicationFactor("store-without-override")) - assertEquals("8", kafkaConfig.getDefaultChangelogStreamReplicationFactor) } @Test