[Spark-2201] Improve FlumeInputDStream's stability and make it scalable

re-submit