[SPARK-3495][SPARK-3496] Backporting fix made in master to branch 1.1

The original PR was #2366 

This backport was non-trivial because Spark 1.1 uses ConnectionManager instead of NioBlockTransferService, which required slight modification to unit tests. Other than that the code is exactly same as in the original PR. Please refer to discussion in the original PR if you have any thoughts.