protected void testFailureDetectorNotification(OperationHelper opHelper, SocketNetworkClient networkClient,
      List<ReplicaId> failedReplicaIds, BlobId blobId, AtomicInteger successfulResponseCount,
      AtomicBoolean invalidResponse, int indexToFail) throws Exception {
    failedReplicaIds.clear();
    successfulResponseCount.set(0);
    invalidResponse.set(false);
    mockSelectorState.set(MockSelectorState.Good);
    FutureResult futureResult = opHelper.submitOperation(blobId);
    int requestParallelism = opHelper.requestParallelism;
    List<RequestInfo> allRequests = new ArrayList<>();
    Set<Integer> allDropped = new HashSet<>();
    long loopStartTimeMs = SystemTime.getInstance().milliseconds();
    while (allRequests.size() < requestParallelism) {
      if (loopStartTimeMs + AWAIT_TIMEOUT_MS < SystemTime.getInstance().milliseconds()) {
        Assert.fail("Waited too long for requests.");
      }
      opHelper.pollOpManager(allRequests, allDropped);
    }
    ReplicaId replicaIdToFail = indexToFail == -1 ? null : allRequests.get(indexToFail).getReplicaId();
    for (RequestInfo requestInfo : allRequests) {
      ResponseInfo responseInfo;
      if (replicaIdToFail != null && replicaIdToFail.equals(requestInfo.getReplicaId())) {
        responseInfo = new ResponseInfo(requestInfo, NetworkClientErrorCode.NetworkError, null);
        requestInfo.getRequest().release();
      } else {
        List<RequestInfo> requestInfoListToSend = new ArrayList<>();
        requestInfoListToSend.add(requestInfo);
        List<ResponseInfo> responseInfoList;
        loopStartTimeMs = SystemTime.getInstance().milliseconds();
        do {
          if (loopStartTimeMs + AWAIT_TIMEOUT_MS < SystemTime.getInstance().milliseconds()) {
            Assert.fail("Waited too long for the response.");
          }
          responseInfoList = networkClient.sendAndPoll(requestInfoListToSend, Collections.emptySet(), 10);
          requestInfoListToSend.clear();
        } while (responseInfoList.size() == 0);
        responseInfo = responseInfoList.get(0);
      }
      opHelper.handleResponse(responseInfo);
      responseInfo.release();
    }
    // Poll once again so that the operation gets a chance to complete.
    allRequests.clear();
    if (testEncryption) {
      opHelper.awaitOpCompletionOrTimeOut(futureResult);
    } else {
      opHelper.pollOpManager(allRequests, allDropped);
    }
    futureResult.get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
    Assert.assertEquals(0, allDropped.size());
    if (indexToFail == -1) {
      Assert.assertEquals("Successful notification should have arrived for replicas that were up",
          opHelper.requestParallelism, successfulResponseCount.get());
      Assert.assertEquals("Failure detector should not have been notified", 0, failedReplicaIds.size());
      Assert.assertFalse("There should be no notifications of any other kind", invalidResponse.get());
    } else {
      Assert.assertEquals("Failure detector should have been notified", 1, failedReplicaIds.size());
      Assert.assertEquals("Failed notification should have arrived for the failed replica", replicaIdToFail,
          failedReplicaIds.get(0));
      Assert.assertEquals("Successful notification should have arrived for replicas that were up",
          opHelper.requestParallelism - 1, successfulResponseCount.get());
      Assert.assertFalse("There should be no notifications of any other kind", invalidResponse.get());
    }
  }
Test that failure detector is correctly notified for all responses regardless of the order in which successful
and failed responses arrive.
@param opHelper the {@link OperationHelper}
@param networkClient the {@link SocketNetworkClient}
@param failedReplicaIds the list that will contain all the replicas for which failure was notified.
@param blobId the id of the blob to get/delete. For puts, this will be null.
@param successfulResponseCount the AtomicInteger that will contain the count of replicas for which success was
                               notified.
@param invalidResponse the AtomicBoolean that will contain whether an unexpected failure was notified.
@param indexToFail if greater than 0, the index representing which response for which failure is to be simulated.
                   For example, if index is 0, then the first response will be failed.
                   If the index is -1, no responses will be failed, and successful responses will be returned to
                   the operation managers.
ambry-router/src/test/java/com/github/ambry/router/NonBlockingRouterTestBase.java