Samza Fluent: NPE is streams.id.samza.system is missing

org.apache.samza.SamzaException: Failed to run application
	at org.apache.samza.runtime.LocalApplicationRunner.run(LocalApplicationRunner.java:136)
	at com.linkedin.beam.runners.samza.runtime.fluent.FluentRuntime$RunnerTask.run(FluentRuntime.java:114)
	... 1 more
Caused by: java.lang.NullPointerException
	at org.apache.samza.system.StreamSpec.validateLogicalIdentifier(StreamSpec.java:200)
	at org.apache.samza.system.StreamSpec.<init>(StreamSpec.java:141)
	at org.apache.samza.system.StreamSpec.<init>(StreamSpec.java:120)
	at org.apache.samza.runtime.AbstractApplicationRunner.getStreamSpec(AbstractApplicationRunner.java:94)
	at org.apache.samza.runtime.AbstractApplicationRunner.getStreamSpec(AbstractApplicationRunner.java:76)
	at org.apache.samza.runtime.AbstractApplicationRunner.getStreamSpec(AbstractApplicationRunner.java:51)
	at org.apache.samza.operators.StreamGraphImpl.getInputStream(StreamGraphImpl.java:65)
	at com.linkedin.beam.runners.samza.runtime.fluent.translator.TranslationContext.getInputStream(TranslationContext.java:23)
	at com.linkedin.beam.runners.samza.runtime.fluent.translator.SourceDataSetTranslator.translate(SourceDataSetTranslator.java:9)
	at com.linkedin.beam.runners.samza.runtime.fluent.translator.SourceDataSetTranslator.translate(SourceDataSetTranslator.java:6)
	at com.linkedin.beam.runners.samza.runtime.fluent.translator.FluentTranslator.applyTranslator(FluentTranslator.java:64)
	at com.linkedin.beam.runners.samza.runtime.fluent.translator.FluentTranslator.access$200(FluentTranslator.java:18)
	at com.linkedin.beam.runners.samza.runtime.fluent.translator.FluentTranslator$1.visitDataSet(FluentTranslator.java:54)
	at com.linkedin.beam.runners.samza.ir.translator.IrGraph.visitTopologically(IrGraph.java:97)
	at com.linkedin.beam.runners.samza.runtime.fluent.translator.FluentTranslator.run(FluentTranslator.java:47)
	at com.linkedin.beam.runners.samza.runtime.fluent.translator.FluentTranslator.translate(FluentTranslator.java:43)
	at com.linkedin.beam.runners.samza.runtime.fluent.FluentRuntime$1.init(FluentRuntime.java:62)
	at org.apache.samza.runtime.AbstractApplicationRunner.getExecutionPlan(AbstractApplicationRunner.java:100)
	at org.apache.samza.runtime.LocalApplicationRunner.run(LocalApplicationRunner.java:112)
	... 2 more
Attachments