@Test
  public void multipleResponsesInFlightTest() throws Exception {
    ExecutorService executorService = Executors.newSingleThreadExecutor();
    AsyncRequestResponseHandler responseHandler = getAsyncRequestResponseHandler(0);
    responseHandler.start();
    try {
      final int EXPECTED_QUEUE_SIZE = 5;

      // test for duplicate request submission
      // submit a few responses that halt on read
      CountDownLatch releaseRead = new CountDownLatch(1);
      byte[] data = null;
      RestRequest restRequest = null;
      MockRestResponseChannel restResponseChannel = null;
      ReadableStreamChannel response = null;
      for (int i = 0; i < EXPECTED_QUEUE_SIZE; i++) {
        data = TestUtils.getRandomBytes(32);
        response = new HaltingRSC(ByteBuffer.wrap(data), releaseRead, executorService);
        restRequest = createRestRequest(RestMethod.GET, "/", null, null);
        restRequest.getMetricsTracker().scalingMetricsTracker.markRequestReceived();
        restResponseChannel = new MockRestResponseChannel(restRequest);
        responseHandler.handleResponse(restRequest, restResponseChannel, response, null);
      }
      assertEquals("Response set size unexpected", EXPECTED_QUEUE_SIZE, responseHandler.getResponseSetSize());
      // attach a listener to the last MockRestResponseChannel
      EventMonitor<MockRestResponseChannel.Event> eventMonitor =
          new EventMonitor<MockRestResponseChannel.Event>(MockRestResponseChannel.Event.OnRequestComplete);
      restResponseChannel.addListener(eventMonitor);
      // we try to re submit the last response. This should fail.
      try {
        responseHandler.handleResponse(restRequest, restResponseChannel, response, null);
        fail("Trying to resubmit a response that is already in flight should have failed");
      } catch (RestServiceException e) {
        assertEquals("Unexpected error code", RestServiceErrorCode.RequestResponseQueuingFailure, e.getErrorCode());
      }
      assertEquals("Response set size unexpected", EXPECTED_QUEUE_SIZE, responseHandler.getResponseSetSize());

      releaseRead.countDown();
      if (!eventMonitor.awaitEvent(1, TimeUnit.SECONDS)) {
        fail("awaitResponse took too long. There might be a problem or the timeout may need to be increased");
      }
      // compare the output. We care only about the last one because we want to make sure the duplicate submission
      // did not mess with the original
      assertArrayEquals("Unexpected data in the response", data, restResponseChannel.getResponseBody());

      // test for shutdown when responses are still in progress
      releaseRead = new CountDownLatch(1);
      List<MockRestResponseChannel> responseChannels = new ArrayList<MockRestResponseChannel>(EXPECTED_QUEUE_SIZE);
      for (int i = 0; i < EXPECTED_QUEUE_SIZE; i++) {
        response = new HaltingRSC(ByteBuffer.allocate(0), releaseRead, executorService);
        restRequest = createRestRequest(RestMethod.GET, "/", null, null);
        restRequest.getMetricsTracker().scalingMetricsTracker.markRequestReceived();
        restResponseChannel = new MockRestResponseChannel(restRequest);
        responseHandler.handleResponse(restRequest, restResponseChannel, response, null);
        responseChannels.add(restResponseChannel);
      }
      assertEquals("Response set size unexpected", EXPECTED_QUEUE_SIZE, responseHandler.getResponseSetSize());
      responseHandler.shutdown();
      assertEquals("Response set size unexpected", 0, responseHandler.getResponseSetSize());
      releaseRead.countDown();
      // all of the responses should have exceptions
      for (MockRestResponseChannel channel : responseChannels) {
        assertNotNull("There should be an exception", channel.getException());
        if (channel.getException() instanceof RestServiceException) {
          RestServiceException rse = (RestServiceException) channel.getException();
          assertEquals("Unexpected RestServiceErrorCode", RestServiceErrorCode.ServiceUnavailable, rse.getErrorCode());
        } else {
          throw channel.getException();
        }
      }
    } finally {
      responseHandler.shutdown();
      executorService.shutdown();
    }
  }
Tests various functions when multiple responses are in flight.
@throws Exception
ambry-rest/src/test/java/com/github/ambry/rest/AsyncRequestResponseHandlerTest.java