[SAMZA-432] Throw a better exception when developers send to an undefined system

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 0.8.0
• Fix Version/s: 0.8.0
• Component/s: container
• Labels: newbie
• Assignee: Chris Riccomini
• Reporter: Chris Riccomini
• Created: 15/Oct/14 00:53
• Updated: 15/Oct/14 20:14
• Resolved: 15/Oct/14 20:14

Description

In SystemProducers.send, we currently do this:

      producers(envelope.getSystemStream.getSystem).send(source, bytesEnvelope.get)

If a developer calls collector.send() with a system that's not defined in their job's config, they get:

Exception in thread "main" java.util.NoSuchElementException: key not found: undefined-stream
	at scala.collection.MapLike$class.default(MapLike.scala:228)
	at scala.collection.AbstractMap.default(Map.scala:58)
	at scala.collection.MapLike$class.apply(MapLike.scala:141)
	at scala.collection.AbstractMap.apply(Map.scala:58)
	at org.apache.samza.system.SystemProducers.send(SystemProducers.scala:65)
	at org.apache.samza.container.TaskInstance$$anonfun$send$2.apply(TaskInstance.scala:170)
	at org.apache.samza.container.TaskInstance$$anonfun$send$2.apply(TaskInstance.scala:170)
	at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
	at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
	at org.apache.samza.container.TaskInstance.send(TaskInstance.scala:170)
	at org.apache.samza.container.RunLoop$$anonfun$send$2.apply(RunLoop.scala:116)
	at org.apache.samza.container.RunLoop$$anonfun$send$2.apply(RunLoop.scala:116)
	at scala.collection.Iterator$class.foreach(Iterator.scala:727)
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1157)
	at scala.collection.MapLike$DefaultValuesIterable.foreach(MapLike.scala:206)
	at org.apache.samza.container.RunLoop.send(RunLoop.scala:116)
	at org.apache.samza.container.RunLoop.run(RunLoop.scala:59)
	at org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:504)
	at org.apache.samza.container.SamzaContainer$.main(SamzaContainer.scala:81)
	at org.apache.samza.container.SamzaContainer.main(SamzaContainer.scala)

NOTE: this stack trace is from an 0.7.0 job, but it's basically the same on master

We should update the SystemProducers.send method to do a .getOrElse, and throw a SamzaException with a more clear warning.