@Test
  public void testResponseHandling() throws Exception {
    try {
      Properties props = getNonBlockingRouterProperties(localDcName);
      VerifiableProperties verifiableProperties = new VerifiableProperties((props));
      setOperationParams();
      final List<ReplicaId> failedReplicaIds = new ArrayList<>();
      final AtomicInteger successfulResponseCount = new AtomicInteger(0);
      final AtomicBoolean invalidResponse = new AtomicBoolean(false);
      ResponseHandler mockResponseHandler = new ResponseHandler(mockClusterMap) {
        @Override
        public void onEvent(ReplicaId replicaId, Object e) {
          if (e instanceof ServerErrorCode) {
            if (e == ServerErrorCode.NoError) {
              successfulResponseCount.incrementAndGet();
            } else {
              invalidResponse.set(true);
            }
          } else {
            failedReplicaIds.add(replicaId);
          }
        }
      };

      // Instantiate a router just to put a blob successfully.
      MockServerLayout mockServerLayout = new MockServerLayout(mockClusterMap);
      setRouter(props, mockServerLayout, new LoggingNotificationSystem());
      setOperationParams();

      // More extensive test for puts present elsewhere - these statements are here just to exercise the flow within the
      // NonBlockingRouter class, and to ensure that operations submitted to a router eventually completes.
      String blobIdStr =
          router.putBlob(putBlobProperties, putUserMetadata, putChannel, new PutBlobOptionsBuilder().build()).get();
      BlobId blobId = RouterUtils.getBlobIdFromString(blobIdStr, mockClusterMap);
      router.close();
      for (MockServer mockServer : mockServerLayout.getMockServers()) {
        mockServer.setServerErrorForAllRequests(ServerErrorCode.NoError);
      }

      SocketNetworkClient networkClient =
          new MockNetworkClientFactory(verifiableProperties, mockSelectorState, MAX_PORTS_PLAIN_TEXT, MAX_PORTS_SSL,
              CHECKOUT_TIMEOUT_MS, mockServerLayout, mockTime).getNetworkClient();
      cryptoJobHandler = new CryptoJobHandler(CryptoJobHandlerTest.DEFAULT_THREAD_COUNT);
      KeyManagementService localKMS =
          new MockKeyManagementService(new KMSConfig(verifiableProperties), singleKeyForKMS);
      putManager = new PutManager(mockClusterMap, mockResponseHandler, new LoggingNotificationSystem(),
          new RouterConfig(verifiableProperties), new NonBlockingRouterMetrics(mockClusterMap, null),
          new RouterCallback(networkClient, new ArrayList<>()), "0", localKMS, cryptoService, cryptoJobHandler,
          accountService, mockTime, MockClusterMap.DEFAULT_PARTITION_CLASS, router);
      OperationHelper opHelper = new OperationHelper(OperationType.PUT);
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, null, successfulResponseCount,
          invalidResponse, -1);
      // Test that if a failed response comes before the operation is completed, failure detector is notified.
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, null, successfulResponseCount,
          invalidResponse, 0);
      // Test that if a failed response comes after the operation is completed, failure detector is notified.
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, null, successfulResponseCount,
          invalidResponse, PUT_REQUEST_PARALLELISM - 1);
      testNoResponseNoNotification(opHelper, failedReplicaIds, null, successfulResponseCount, invalidResponse);
      testResponseDeserializationError(opHelper, networkClient, null);

      opHelper = new OperationHelper(OperationType.GET);
      getManager = new GetManager(mockClusterMap, mockResponseHandler, new RouterConfig(verifiableProperties),
          new NonBlockingRouterMetrics(mockClusterMap, null),
          new RouterCallback(networkClient, new ArrayList<BackgroundDeleteRequest>()), localKMS, cryptoService,
          cryptoJobHandler, mockTime, null, router);
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, blobId, successfulResponseCount,
          invalidResponse, -1);
      // Test that if a failed response comes before the operation is completed, failure detector is notified.
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, blobId, successfulResponseCount,
          invalidResponse, 0);
      // Test that if a failed response comes after the operation is completed, failure detector is notified.
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, blobId, successfulResponseCount,
          invalidResponse, GET_REQUEST_PARALLELISM - 1);
      testNoResponseNoNotification(opHelper, failedReplicaIds, blobId, successfulResponseCount, invalidResponse);
      testResponseDeserializationError(opHelper, networkClient, blobId);

      opHelper = new OperationHelper(OperationType.DELETE);
      deleteManager =
          new DeleteManager(mockClusterMap, mockResponseHandler, accountService, new LoggingNotificationSystem(),
              new RouterConfig(verifiableProperties), new NonBlockingRouterMetrics(mockClusterMap, null),
              new RouterCallback(null, new ArrayList<BackgroundDeleteRequest>()), mockTime, router);
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, blobId, successfulResponseCount,
          invalidResponse, -1);
      // Test that if a failed response comes before the operation is completed, failure detector is notified.
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, blobId, successfulResponseCount,
          invalidResponse, 0);
      // Test that if a failed response comes after the operation is completed, failure detector is notified.
      testFailureDetectorNotification(opHelper, networkClient, failedReplicaIds, blobId, successfulResponseCount,
          invalidResponse, DELETE_REQUEST_PARALLELISM - 1);
      testNoResponseNoNotification(opHelper, failedReplicaIds, blobId, successfulResponseCount, invalidResponse);
      testResponseDeserializationError(opHelper, networkClient, blobId);
    } finally {
      if (putManager != null) {
        putManager.close();
      }
      if (getManager != null) {
        getManager.close();
      }
      if (deleteManager != null) {
        deleteManager.close();
      }
    }
  }
Response handling related tests for all operation managers.
ambry-router/src/test/java/com/github/ambry/router/NonBlockingRouterTest.java