SPARK-1729. Make Flume pull data from source, rather than the current pu...

...sh model

Currently Spark uses Flume's internal Avro Protocol to ingest data from Flume. If the executor running the
receiver fails, it currently has to be restarted on the same node to be able to receive data.

This commit adds a new Sink which can be deployed to a Flume agent. This sink can be polled by a new
DStream that is also included in this commit. This model ensures that data can be pulled into Spark from
Flume even if the receiver is restarted on a new node. This also allows the receiver to receive data on
multiple threads for better performance.
But the spark streaming side would not know anything about why it failed. Was it a failure to create transaction, or some exception in get data from upstream. 

So does it make sense to have something like `class ErrorEventBatch(reason: String) extend EventBatch` which is created and returned in case of an exception (the reason having the exception string)? Then on the spark streaming we will have more information to debug, isnt it?