Samza 1214: Allow users to set a default replication.factor for intermediate topics

* Add a new "systems.sysName.default.stream.*" config structure that allows users to set system-wide defaults for streams.
* More thorough testing of system defaults and stream defaults
* Removed the old migration config from the config table since there's no code to support it.
* Moved 2 kafka-specific config accessors out of JobConfig and into KafkaConfig
* Removed duplicate impl of getChangelogStream()

Author: Jacob Maes <jmaes@linkedin.com>

Reviewers: Jagadish <jvenkatr@linkedin.com>

Closes apache#141 from jmakes/samza-1214

Author: Jacob Maes

docs/learn/documentation/versioned/jobs/configuration-table.html

@@ -720,8 +720,19 @@ <h1>Samza Configuration Reference</h1>
                 </tr>
 
                 <tr>
-                    <td class="property" id="systems-samza-key-serde">systems.<span class="system">system-name</span>.<br>samza.key.serde</td>
-                    <td class="default" rowspan="2"></td>
+                    <td class="property" id="systems-default-stream">systems.<span class="system">system-name</span>.<br>default.stream.*</td>
+                    <td class="default"></td>
+                    <td class="description">
+                        A set of default properties for any stream associated with the system. For example, if
+                        "systems.kafka-system.default.stream.replication.factor"=2 was configured, then every Kafka stream
+                        created on the kafka-system will have a replication factor of 2 unless the property is explicitly
+                        overridden at the stream scope using <a href="#streams-properties">streams properties</a>.
+                    </td>
+                </tr>
+
+                <tr>
+                    <td class="property" id="systems-samza-key-serde">systems.<span class="system">system-name</span>.<br>default.stream.samza.key.serde</td>
+                    <td class="default"></td>
                     <td class="description">
                         The <a href="../container/serialization.html">serde</a> which will be used to deserialize the
                         <em>key</em> of messages on input streams, and to serialize the <em>key</em> of messages on
@@ -734,17 +745,10 @@ <h1>Samza Configuration Reference</h1>
                         the task and the output stream producer.
                     </td>
                 </tr>
-                <tr>
-                    <td class="property">systems.<span class="system">system-name</span>.<br>streams.<span class="stream">stream-name</span>.<br>samza.key.serde</td>
-                    <td class="description">
-                        This is deprecated in favor of <a href="#streams-samza-key-serde" class="property">
-                        streams.<span class="stream">stream-id</span>.samza.key.serde</a>.
-                    </td>
-                </tr>
 
                 <tr>
-                    <td class="property" id="systems-samza-msg-serde">systems.<span class="system">system-name</span>.<br>samza.msg.serde</td>
-                    <td class="default" rowspan="2"></td>
+                    <td class="property" id="systems-samza-msg-serde">systems.<span class="system">system-name</span>.<br>default.stream.samza.msg.serde</td>
+                    <td class="default"></td>
                     <td class="description">
                         The <a href="../container/serialization.html">serde</a> which will be used to deserialize the
                         <em>value</em> of messages on input streams, and to serialize the <em>value</em> of messages on
@@ -757,17 +761,10 @@ <h1>Samza Configuration Reference</h1>
                         the task and the output stream producer.
                     </td>
                 </tr>
-                <tr>
-                    <td class="property">systems.<span class="system">system-name</span>.<br>streams.<span class="stream">stream-name</span>.<br>samza.msg.serde</td>
-                    <td class="description">
-                        This is deprecated in favor of <a href="#streams-samza-msg-serde" class="property">
-                        streams.<span class="stream">stream-id</span>.samza.msg.serde</a>.
-                    </td>
-                </tr>
 
                 <tr>
-                    <td class="property" id="systems-samza-offset-default">systems.<span class="system">system-name</span>.<br>samza.offset.default</td>
-                    <td class="default" rowspan="2">upcoming</td>
+                    <td class="property" id="systems-samza-offset-default">systems.<span class="system">system-name</span>.<br>default.stream.samza.offset.default</td>
+                    <td class="default">upcoming</td>
                     <td class="description">
                         If a container starts up without a <a href="../container/checkpointing.html">checkpoint</a>,
                         this property determines where in the input stream we should start consuming. The value must be an
@@ -786,13 +783,6 @@ <h1>Samza Configuration Reference</h1>
                         If both are defined, the stream-level definition takes precedence.
                     </td>
                 </tr>
-                <tr>
-                    <td class="property">systems.<span class="system">system-name</span>.<br>streams.<span class="stream">stream-name</span>.<br>samza.offset.default</td>
-                    <td class="description">
-                        This is deprecated in favor of <a href="#streams-samza-offset-default" class="property">
-                        streams.<span class="stream">stream-id</span>.samza.offset.default</a>.
-                    </td>
-                </tr>
 
                 <tr>
                     <td class="property" id="task-consumer-batch-size">task.consumer.batch.size</td>
@@ -1352,7 +1342,7 @@ <h1>Samza Configuration Reference</h1>
                         This property defines a store, Samza's mechanism for efficient
                         <a href="../container/state-management.html">stateful stream processing</a>. You can give a
                         store any <span class="store">store-name</span> except <em>default</em> (the <span class="store">store-name</span>
-                        <em>default</em> is reserved for defining default store parameters), and use that name to get a 
+                        <em>default</em> is reserved for defining default store parameters), and use that name to get a
                         reference to the store in your stream task (call
                         <a href="../api/javadocs/org/apache/samza/task/TaskContext.html#getStore(java.lang.String)">TaskContext.getStore()</a>
                         in your task's
@@ -1983,29 +1973,6 @@ <h1>Samza Configuration Reference</h1>
                     <td class="default"></td>
                     <td class="description">Staging directory for storing partition description. By default (if not set by users) the value is inherited from "yarn.job.staging.directory" internally. The default value is typically good enough unless you want explicitly use a separate location.</td>
                 </tr>
-
-                <tr>
-                    <th colspan="3" class="section" id="task-migration">
-                        Migrating from Samza 0.9.1 to 0.10.0<br>
-                        <span class="subtitle">
-                            (This section applies if you are upgrading from Samza 0.9.1 to 0.10.0 and have set
-                            <a href="#task-checkpoint-factory" class="property">task.checkpoint.factory</a> to anything <b> other than </b>
-                            <code>org.apache.samza.checkpoint.file.FileSystemCheckpointManagerFactory</code>)
-                        </span>
-                    </th>
-                </tr>
-
-                <tr>
-                    <td class="property" id="task-checkpoint-skip-migration">task.checkpoint.skip-migration</td>
-                    <td class="default">false</td>
-                    <td class="description">
-                        When migrating from 0.9.1 to 0.10.0, the taskName-to-changelog partition mapping was moved from the checkpoint stream to the coordinator stream. <br />
-                        If you are using a checkpoint manager other than kafka
-                        (<code>org.apache.samza.checkpoint.kafka.KafkaCheckpointManagerFactory</code>), you have to
-                        manually migrate taskName-to-changelog partition mapping to the coordinator stream. <br />
-                        This can be achieved with the assistance of the <code>checkpoint-tool.sh</code>.
-                    </td>
-                </tr>
             </tbody>
         </table>
     </body>

samza-api/src/main/java/org/apache/samza/config/MapConfig.java

@@ -19,6 +19,7 @@
 
 package org.apache.samza.config;
 
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
@@ -41,11 +42,15 @@ public MapConfig(Map<String, String> map) {
   }
 
   public MapConfig(List<Map<String, String>> maps) {
-    this.map = new HashMap<String, String>();
+    this.map = new HashMap<>();
     for (Map<String, String> m: maps)
       this.map.putAll(m);
   }
 
+  public MapConfig(Map<String, String>... maps) {
+    this(Arrays.asList(maps));
+  }
+
   public String get(Object k) {
     return map.get(k);
   }

samza-core/src/main/java/org/apache/samza/config/JavaStorageConfig.java

@@ -58,7 +58,7 @@ public String getChangelogStream(String storeName) {
     // If this config only specifies <astream> and there is a value in job.changelog.system=<asystem> -
     // these values will be combined into <asystem>.<astream>
     String systemStream = get(String.format(CHANGELOG_STREAM, storeName), null);
-    String changelogSystem = get(CHANGELOG_SYSTEM, null);
+    String changelogSystem = getChangelogSystem();
 
     String systemStreamRes;
     if (systemStream != null  && !systemStream.contains(".")) {
@@ -85,4 +85,21 @@ public String getStorageKeySerde(String storeName) {
   public String getStorageMsgSerde(String storeName) {
     return get(String.format(MSG_SERDE, storeName), null);
   }
+
+  /**
+   * Gets the System to use for reading/writing checkpoints. Uses the following precedence.
+   *
+   * 1. If job.changelog.system is defined, that value is used.
+   * 2. If job.default.system is defined, that value is used.
+   * 3. null
+   *
+   * Note: Changelogs can be defined using
+   * stores.storeName.changelog=systemName.streamName  or
+   * stores.storeName.changelog=streamName
+   *
+   * If the former syntax is used, that system name will still be honored. For the latter syntax, this method is used.
+   */
+  public String getChangelogSystem() {
+    return get(CHANGELOG_SYSTEM,  get(JobConfig.JOB_DEFAULT_SYSTEM(), null));
+  }
 }

samza-core/src/main/java/org/apache/samza/config/JavaSystemConfig.java

@@ -35,7 +35,8 @@
 public class JavaSystemConfig extends MapConfig {
   private static final String SYSTEM_PREFIX = "systems.";
   private static final String SYSTEM_FACTORY_SUFFIX = ".samza.factory";
-  private static final String SYSTEM_FACTORY = "systems.%s.samza.factory";
+  private static final String SYSTEM_FACTORY_FORMAT = SYSTEM_PREFIX + "%s" + SYSTEM_FACTORY_SUFFIX;
+  private static final String SYSTEM_DEFAULT_STREAMS_PREFIX = SYSTEM_PREFIX + "%s" + ".default.stream.";
   private static final String EMPTY = "";
 
   public JavaSystemConfig(Config config) {
@@ -46,7 +47,7 @@ public String getSystemFactory(String name) {
     if (name == null) {
       return null;
     }
-    String systemFactory = String.format(SYSTEM_FACTORY, name);
+    String systemFactory = String.format(SYSTEM_FACTORY_FORMAT, name);
     return get(systemFactory, null);
   }
 
@@ -99,4 +100,11 @@ public Map<String, SystemFactory> getSystemFactories() {
 
     return systemFactories;
   }
+
+  /**
+   * Gets the system-wide defaults for streams.
+   */
+  public Config getDefaultStreamProperties(String systemName) {
+    return subset(String.format(SYSTEM_DEFAULT_STREAMS_PREFIX, systemName), true);
+  }
 }

samza-core/src/main/scala/org/apache/samza/config/JobConfig.scala

@@ -23,7 +23,6 @@ package org.apache.samza.config
 import java.io.File
 
 import org.apache.samza.container.grouper.stream.GroupByPartitionFactory
-import org.apache.samza.system.{RegexSystemStreamPartitionMatcher, SystemStreamPartitionMatcher}
 import org.apache.samza.util.Logging
 
 object JobConfig {
@@ -47,8 +46,6 @@ object JobConfig {
   val JOB_CONTAINER_COUNT = "job.container.count"
   val JOB_CONTAINER_THREAD_POOL_SIZE = "job.container.thread.pool.size"
   val JOB_CONTAINER_SINGLE_THREAD_MODE = "job.container.single.thread.mode"
-  val JOB_REPLICATION_FACTOR = "job.coordinator.replication.factor"
-  val JOB_SEGMENT_BYTES = "job.coordinator.segment.bytes"
   val JOB_INTERMEDIATE_STREAM_PARTITIONS = "job.intermediate.stream.partitions"
 
   val SSP_GROUPER_FACTORY = "job.systemstreampartition.grouper.factory"
@@ -104,8 +101,22 @@ object JobConfig {
 class JobConfig(config: Config) extends ScalaMapConfig(config) with Logging {
   def getName = getOption(JobConfig.JOB_NAME)
 
-  def getCoordinatorSystemName = getOption(JobConfig.JOB_COORDINATOR_SYSTEM).getOrElse(
-      throw new ConfigException("Missing job.coordinator.system configuration. Cannot proceed with job execution."))
+  def getCoordinatorSystemName = {
+    val system = getCoordinatorSystemNameOrNull
+    if (system == null) {
+      throw new ConfigException("Missing job.coordinator.system configuration. Cannot proceed with job execution.")
+    }
+    system
+  }
+
+  /**
+    * Gets the System to use for reading/writing the coordinator stream. Uses the following precedence.
+    *
+    * 1. If job.coordinator.system is defined, that value is used.
+    * 2. If job.default.system is defined, that value is used.
+    * 3. None
+    */
+  def getCoordinatorSystemNameOrNull =  getOption(JobConfig.JOB_COORDINATOR_SYSTEM).getOrElse(getDefaultSystem.orNull)
 
   def getDefaultSystem = getOption(JobConfig.JOB_DEFAULT_SYSTEM)
 
@@ -144,31 +155,6 @@ class JobConfig(config: Config) extends ScalaMapConfig(config) with Logging {
 
   def getSecurityManagerFactory = getOption(JobConfig.JOB_SECURITY_MANAGER_FACTORY)
 
-  val CHECKPOINT_SEGMENT_BYTES = "task.checkpoint.segment.bytes"
-  val CHECKPOINT_REPLICATION_FACTOR = "task.checkpoint.replication.factor"
-
-  def getCoordinatorReplicationFactor = getOption(JobConfig.JOB_REPLICATION_FACTOR) match {
-    case Some(rplFactor) => rplFactor
-    case _ =>
-      getOption(CHECKPOINT_REPLICATION_FACTOR) match {
-        case Some(rplFactor) =>
-          info("%s was not found. Using %s=%s for coordinator stream" format (JobConfig.JOB_REPLICATION_FACTOR, CHECKPOINT_REPLICATION_FACTOR, rplFactor))
-          rplFactor
-        case _ => "3"
-      }
-  }
-
-  def getCoordinatorSegmentBytes = getOption(JobConfig.JOB_SEGMENT_BYTES) match {
-    case Some(segBytes) => segBytes
-    case _ =>
-      getOption(CHECKPOINT_SEGMENT_BYTES) match {
-        case Some(segBytes) =>
-          info("%s was not found. Using %s=%s for coordinator stream" format (JobConfig.JOB_SEGMENT_BYTES, CHECKPOINT_SEGMENT_BYTES, segBytes))
-          segBytes
-        case _ => "26214400"
-      }
-  }
-
   def getSSPMatcherClass = getOption(JobConfig.SSP_MATCHER_CLASS)
 
   def getSSPMatcherConfigRegex = getExcept(JobConfig.SSP_MATCHER_CONFIG_REGEX)

samza-core/src/main/scala/org/apache/samza/config/StorageConfig.scala

@@ -46,24 +46,8 @@ class StorageConfig(config: Config) extends ScalaMapConfig(config) with Logging
   def getStorageMsgSerde(name: String) = getOption(StorageConfig.MSG_SERDE format name)
 
   def getChangelogStream(name: String) = {
-    // If the config specifies 'stores.<storename>.changelog' as '<system>.<stream>' combination - it will take precedence.
-    // If this config only specifies <astream> and there is a value in job.changelog.system=<asystem> -
-    // these values will be combined into <asystem>.<astream>
-    val systemStream = getOption(CHANGELOG_STREAM format name)
-    val changelogSystem = getOption(CHANGELOG_SYSTEM)
-    val systemStreamRes =
-      if ( systemStream.isDefined  && ! systemStream.getOrElse("").contains('.')) {
-        // contains only stream name
-        if (changelogSystem.isDefined) {
-          Some(changelogSystem.get + "." + systemStream.get)
-        }
-        else {
-          throw new SamzaException("changelog system is not defined:" + systemStream.get)
-        }
-      } else {
-        systemStream
-      }
-    systemStreamRes
+    val javaStorageConfig = new JavaStorageConfig(config)
+    Option(javaStorageConfig.getChangelogStream(name))
   }
 
   def getChangeLogDeleteRetentionInMs(storeName: String) = {

samza-core/src/main/scala/org/apache/samza/config/StreamConfig.scala

@@ -81,11 +81,14 @@ class StreamConfig(config: Config) extends ScalaMapConfig(config) with Logging {
    * Returns a list of all SystemStreams that have a serde defined from the config file.
    */
   def getSerdeStreams(systemName: String) = {
+    val defaultStreamProperties = new JavaSystemConfig(config).getDefaultStreamProperties(systemName)
+    val hasSystemDefaultSerde = defaultStreamProperties.containsKey(StreamConfig.MSG_SERDE) || defaultStreamProperties.containsKey(StreamConfig.KEY_SERDE)
+
     val subConf = config.subset("systems.%s.streams." format systemName, true)
     val legacySystemStreams = subConf
       .asScala
       .keys
-      .filter(k => k.endsWith(StreamConfig.MSG_SERDE) || k.endsWith(StreamConfig.KEY_SERDE))
+      .filter(k => k.endsWith(StreamConfig.MSG_SERDE) || k.endsWith(StreamConfig.KEY_SERDE) || hasSystemDefaultSerde)
       .map(k => {
         val streamName = k.substring(0, k.length - 16 /* .samza.XXX.serde length */ )
         new SystemStream(systemName, streamName)
@@ -94,7 +97,7 @@ class StreamConfig(config: Config) extends ScalaMapConfig(config) with Logging {
     val systemStreams = subset(StreamConfig.STREAMS_PREFIX)
       .asScala
       .keys
-      .filter(k => k.endsWith(StreamConfig.MSG_SERDE) || k.endsWith(StreamConfig.KEY_SERDE))
+      .filter(k => k.endsWith(StreamConfig.MSG_SERDE) || k.endsWith(StreamConfig.KEY_SERDE) || hasSystemDefaultSerde)
       .map(k => k.substring(0, k.length - 16 /* .samza.XXX.serde length */ ))
       .filter(streamId => systemName.equals(getSystem(streamId)))
       .map(streamId => streamIdToSystemStream(streamId)).toSet
@@ -220,10 +223,13 @@ class StreamConfig(config: Config) extends ScalaMapConfig(config) with Logging {
     * @return           the map of properties for the stream
     */
   private def getSystemStreamProperties(systemName: String, streamName: String) = {
-    if (systemName == null || streamName == null) {
+    if (systemName == null) {
       Map()
     }
-    config.subset(StreamConfig.STREAM_PREFIX format(systemName, streamName), true)
+    val systemConfig = new JavaSystemConfig(config);
+    val defaults = systemConfig.getDefaultStreamProperties(systemName);
+    val explicitConfigs = config.subset(StreamConfig.STREAM_PREFIX format(systemName, streamName), true)
+    new MapConfig(defaults, explicitConfigs)
   }
 
   /**