@Override
  public void handleRequest(RestRequest request, RequestContext requestContext, final Callback<RestResponse> callback)
  {
    if (HttpMethod.POST != HttpMethod.valueOf(request.getMethod()))
    {
      _log.error("POST is expected, but " + request.getMethod() + " received");
      callback.onError(RestException.forError(HttpStatus.S_405_METHOD_NOT_ALLOWED.getCode(), "Invalid method"));
      return;
    }

    // Disable server-side latency instrumentation for multiplexed requests
    requestContext.putLocalAttr(TimingContextUtil.TIMINGS_DISABLED_KEY_NAME, true);

    IndividualRequestMap individualRequests;
    try
    {
      individualRequests = extractIndividualRequests(request);
      if (_multiplexerSingletonFilter != null) {
        individualRequests = _multiplexerSingletonFilter.filterRequests(individualRequests);
      }
    }
    catch (RestException e)
    {
      _log.error("Invalid multiplexed request", e);
      callback.onError(e);
      return;
    }
    catch (Exception e)
    {
      _log.error("Invalid multiplexed request", e);
      callback.onError(RestException.forError(HttpStatus.S_400_BAD_REQUEST.getCode(), e));
      return;
    }
    // prepare the map of individual responses to be collected
    final IndividualResponseMap individualResponses = new IndividualResponseMap(individualRequests.size());
    final Map<String, HttpCookie> responseCookies = new HashMap<>();
    // all tasks are Void and side effect based, that will be useful when we add streaming
    Task<?> requestProcessingTask = createParallelRequestsTask(request, requestContext, individualRequests, individualResponses, responseCookies);
    Task<Void> responseAggregationTask = Task.action("send aggregated response", () ->
      {
        RestResponse aggregatedResponse = aggregateResponses(individualResponses, responseCookies);
        callback.onSuccess(aggregatedResponse);
      }
    );
    _engine.run(requestProcessingTask.andThen(responseAggregationTask), MUX_PLAN_CLASS);
  }
@Test(dataProvider = "multiplexerConfigurations")
  public void testMultiplexedSingletonFilterFailures(MultiplexerRunMode multiplexerRunMode) throws Exception
  {
    // This test validates when a failure occurred in MultiplexedSingletonFilter for an individual request, only the individual
    // request should fail. The multiplexed request should still be completed successfully with a 200 status code.

    // Setup mock request handler: make handler return a json that contains the request uri
    // We are using this uri in our mock MultiplexerSingletonFilter.filterIndividualResponse function so that
    // we can simulate different response based on the request.
    SynchronousRequestHandler mockHandler = new SynchronousRequestHandler() {
      @Override
      public RestResponse handleRequestSync(RestRequest request, RequestContext requestContext)
      {
        try
        {
          return fakeIndRestResponse(jsonBodyToByteString(fakeIndividualBody(request.getURI().toString())));
        }
        catch (Exception e)
        {
          throw new RuntimeException(e);
        }
      }
    };

    // Create a mock MultiplexerSingletonFilter to it simulate different type of failures.
    // Failure is simulated are base on the request uri.
    MultiplexerSingletonFilter muxFilterWithSimulatedFailures = new MultiplexerSingletonFilter() {
      @Override
      public IndividualRequest filterIndividualRequest(IndividualRequest request)
      {
        if (request.getRelativeUrl().contains("bad_request"))
        {
          throw new RestLiServiceException(HttpStatus.S_400_BAD_REQUEST, "not found");
        }
        else if (request.getRelativeUrl().contains("error_request"))
        {
          throw new IllegalArgumentException("Something really bad happened in filterIndividualRequest");
        }
        return request;
      }

      @Override
      public IndividualResponse filterIndividualResponse(IndividualResponse response)
      {
        if (response.getStatus() == HttpStatus.S_200_OK.getCode())
        {
          if (response.getBody().data().getString("value").contains("notfound_response"))
          {
            throw new RestLiServiceException(HttpStatus.S_404_NOT_FOUND, "not found");
          }
          else if (response.getBody().data().getString("value").contains("error_response"))
          {
            // simulate an unexpected exception
            throw new UnsupportedOperationException("Something really bad happened in filterIndividualResponse");
          }
        }
        return response;
      }
    };

    // Prepare request to mux handler
    FutureCallback<RestResponse> callback = new FutureCallback<>();
    RequestContext requestContext = new RequestContext();
    Map<String, IndividualRequest> individualRequests = new HashMap<>();
    individualRequests.put("0", fakeIndRequest("/good_request"));
    individualRequests.put("1", fakeIndRequest("/bad_request"));
    individualRequests.put("2", fakeIndRequest("/error_request"));
    individualRequests.put("3", fakeIndRequest("/notfound_response"));
    individualRequests.put("4", fakeIndRequest("/error_response"));
    individualRequests.put("5", fakeIndRequest("/good_request", ImmutableMap.of("6", fakeIndRequest("/bad_request"))));


    RestRequest request = fakeMuxRestRequest(individualRequests);

    // Create mux handler instance
    MultiplexedRequestHandlerImpl multiplexer = createMultiplexer(mockHandler, muxFilterWithSimulatedFailures, Collections.<String>emptySet(), 10, multiplexerRunMode);

    try
    {
      multiplexer.handleRequest(request, requestContext, callback);
    }
    catch (Exception e)
    {
      fail("Multiplexer should not blow up because one of the individual requests failed", e);
    }

    RestResponse muxRestResponse = callback.get();
    // Assert multiplexed request should return a 200 status code
    assertEquals(muxRestResponse.getStatus(), 200, "Failure in indivudal request should not cause the entire multliplexed request to fail");
    MultiplexedResponseContent muxResponseContent = new MultiplexedResponseContent(DataMapConverter.bytesToDataMap(muxRestResponse.getHeaders(), muxRestResponse.getEntity()));

    IndividualResponseMap responses = muxResponseContent.getResponses();
    // Validate the status code for each of the response
    assertEquals(responses.get("0").getStatus().intValue(), 200, "Mux response body is: " +  responses.toString());
    assertEquals(responses.get("1").getStatus().intValue(), 400, "Mux response body is: " + responses.toString());
    assertEquals(responses.get("2").getStatus().intValue(), 500, "Mux response body is: " +  responses.toString());
    assertEquals(responses.get("3").getStatus().intValue(), 404, "Mux response body is: " +  responses.toString());
    assertEquals(responses.get("4").getStatus().intValue(), 500, "Mux response body is: " + responses.toString());
    assertEquals(responses.get("5").getStatus().intValue(), 200, "Mux response body is: " + responses.toString());
    assertEquals(responses.get("6").getStatus().intValue(), 400, "Mux response body is: " + responses.toString());
  }