[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.
There can still be a race condition, I think! Another thread can stay stuck in line 81, while the shutting threads enter critical sections, and marks all the processors for shutdown. Then when shutting thread exits critical section, the other thread (stuck in 81) will enter its ciritical and create a new transaction processor. 

Isnt it?

More fundamentally, why do we even need the activeProcessors? We can added all the processors created to the `sequenceNumberToProcessor` map, even before the processors has been submitted for populating events. At the time of shutdown, we can do `sequenceNumberToProcessor.values.foreach(_.shutdown)` to shutdown all of them. This would be easier to reason with regarding race conditions, because then we dont have to add processor in two different data structures (`activeProcessors` and `sequenceNumberToProcessor`), in two different steps (before populateEvents and after). 

And in that case it become easier to synchronize. Let there be two methods with the following semantics.
```
def createProcessor(): Option[TransactionProcessor] = sequenceNumberToProcessor.synchronized {
    if (!stopped) {
        // create processor, add to the map, and return Some(proc)
    } else None.
}

def removeAndGetProcessor(seqNum: CharSequence): TransactionProcessor = sequenceNumberToProcessor.synchronized  {
    // remove and return within the lock
}

def shutdown(): Unit = {
     sequenceNumberToProcessor.synchronized {
             stopped = true
             // shutdown all the processors
     }
}
```