code: 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();
    }
docs: Fire the failure callback for all outstanding requests. This is called when we have an
  uncaught exception or pre-mature connection termination.