private void failOutstandingRequests ( Throwable cause ) { for ( Map . Entry < StreamChunkId , ChunkReceivedCallback > entry : outstandingFetches . entrySet ( ) ) { try { entry . getValue ( ) . onFailure ( entry . getKey ( ) . chunkIndex , cause ) ; } catch ( Exception e ) { logger . warn ( "ChunkReceivedCallback.onFailure throws exception" , e ) ; } } for ( Map . Entry < Long , RpcResponseCallback > entry : outstandingRpcs . entrySet ( ) ) { try { entry . getValue ( ) . onFailure ( cause ) ; } catch ( Exception e ) { logger . warn ( "RpcResponseCallback.onFailure throws exception" , e ) ; } } for ( Pair < String , StreamCallback > entry : streamCallbacks ) { try { entry . getValue ( ) . onFailure ( entry . getKey ( ) , cause ) ; } catch ( Exception e ) { logger . warn ( "StreamCallback.onFailure throws exception" , e ) ; } } // It's OK if new fetches appear, as they will fail immediately. outstandingFetches . clear ( ) ; outstandingRpcs . clear ( ) ; streamCallbacks . clear ( ) ; }
Fire the failure callback for all outstanding requests . This is called when we have an uncaught exception or pre - mature connection termination .