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.
The rollback logs it as a generic "rolling back" message; it does not log the reason behind the rollback in detail enough. From what I see, there are a number of cases when a rollback can occur
- no data received
- exception in receiving data
- data received, sent to receiver, but nack received
- data received, sent to receiver, but ack not received by timeout
- data receiver, sent to receiver, exception will waiting for ack (may be same as above)

We should be able to very easily identify which of the cases caused the rollback of the transaction. It need not require logging in the rollbackAndClose function, but at least all control paths leading to a failure of the transaction should get logged for easy debugging.