SAMZA-1245: Make stream samza.physical.name config name string public

For certain system such as hdfs, the physical stream name might need to be finalized during the config generation. In order to do that, we will need to expose the stream samza.physical.name config string.

Author: Xinyu Liu <[email protected]>

Reviewers: Jake Maes <[email protected]>

Closes apache#145 from xinyuiscool/SAMZA-1245

Author: Xinyu Liu

samza-core/src/main/java/org/apache/samza/execution/StreamManager.java

@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.apache.samza.SamzaException;
 import org.apache.samza.system.StreamSpec;
 import org.apache.samza.system.SystemAdmin;
 import org.apache.samza.system.SystemStreamMetadata;
@@ -62,6 +63,10 @@ Map<String, Integer> getStreamPartitionCounts(String systemName, Set<String> str
     Map<String, Integer> streamToPartitionCount = new HashMap<>();
 
     SystemAdmin systemAdmin = sysAdmins.get(systemName);
+    if (systemAdmin == null) {
+      throw new SamzaException(String.format("System %s does not exist.", systemName));
+    }
+
     // retrieve the metadata for the streams in this system
     Map<String, SystemStreamMetadata> streamToMetadata = systemAdmin.getSystemStreamMetadata(streamNames);
     // set the partitions of a stream to its StreamEdge

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

@@ -43,7 +43,7 @@ object StreamConfig {
 
   protected val STREAM_ID_PREFIX = STREAMS_PREFIX + "%s."
   protected val SYSTEM_FOR_STREAM_ID = STREAM_ID_PREFIX + SYSTEM
-  protected val PHYSICAL_NAME_FOR_STREAM_ID = STREAM_ID_PREFIX + PHYSICAL_NAME
+  val PHYSICAL_NAME_FOR_STREAM_ID = STREAM_ID_PREFIX + PHYSICAL_NAME
 
   implicit def Config2Stream(config: Config) = new StreamConfig(config)
 }