[SPARK-11762] [network] Account for active streams when couting outstanding requests.

[vanzin]

This way the timeout handling code can correctly close "hung" channels that are
processing streams.

[SparkQA]

Test build #46027 has finished for PR 9747 at commit 4df514c.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):\n * class StreamingListener(object):\n

[vanzin]

/cc @rxin @zsxwing

[zsxwing]

LGTM.

BTW, StreamInterceptor should catch exception from callback.onComplete, onData, onFailure. Otherwise, it may screw up the stream.

[zsxwing]

BTW, StreamInterceptor should catch exception from callback.onComplete, onData, onFailure. Otherwise, it may screw up the stream.

@vanzin could you fix this in a separate PR?

[vanzin]

BTW, StreamInterceptor should catch exception from callback.onComplete, onData, onFailure.

[vanzin]

Oops. Hit the wrong thing.

[vanzin]

BTW, StreamInterceptor should catch exception from callback.onComplete, onData, onFailure. Otherwise, it may screw up the stream.

It's not as simple as catching the exceptions. StreamInterceptor would have to switch to a mode where it ignores all future data until the stream is over, if the channel is to be kept open.

Right now an exception will bubble up and end up handled by TransportChannelHandler, which will close the channel.

[zsxwing]

Right now an exception will bubble up and end up handled by TransportChannelHandler, which will close the channel.

Sounds acceptable since #9530 uses a separate channel to stream files.

[vanzin]

Merging to master.