[SPARK-3154][STREAMING] Make FlumePollingInputDStream shutdown cleaner.

Currently lot of errors get thrown from Avro IPC layer when the dstream
or sink is shutdown. This PR cleans it up. Some refactoring is done in the
receiver code to put all of the RPC code into a single Try and just recover
from that. The sink code has also been cleaned up.
Yes, so does the whole project. The convention seems to be to let it come in from spark-core. I sort of remember that this caused some problem that looked like a Scala compiler problem -- that is, shouldn't be a problem, but was tickling some issue. Let me see if I can dig that out, or whether I'm misremembering.