@Override
  @SuppressWarnings("unchecked")
  public <K, T> void onAllResponsesReceived(Request<T> request, ProtocolVersion protocolVersion,
                                            Map<RequestInfo, Response<T>> successResponses,
                                            Map<RequestInfo, Throwable> failureResponses,
                                            Map<Integer, Set<K>> unmappedKeys,
                                            Callback<Response<T>> callback)
  {
    BatchRequest<T> batchRequest = safeCastRequest(request);
    // initialize an empty dataMap for final response entity
    DataMap gatheredResponseDataMap = initializeResponseContainer();
    // gather success response
    successResponses.forEach((req, response) -> gatherResponse(gatheredResponseDataMap, req, response.getEntity()));
    // gather failure response
    failureResponses.forEach((req, e) ->
            {
              Set<K> failedKeys = (Set<K>)((BatchRequest<T>)req.getRequest()).getObjectIds();
              gatherException(gatheredResponseDataMap, failedKeys, e, protocolVersion);
            });
    // gather unmapped keys
    if (unmappedKeys != null && !unmappedKeys.isEmpty())
    {
      Set<K> unmapped = unmappedKeys.values().stream().flatMap(Collection::stream).collect(Collectors.toSet());
      gatherException(gatheredResponseDataMap, unmapped,
              new RestLiScatterGatherException("Unable to find a host for keys :" + unmapped),
              protocolVersion);
    }
    T gatheredResponse = constructResponseFromDataMap(batchRequest, protocolVersion, gatheredResponseDataMap);
    if (!successResponses.isEmpty())
    {
      Response<T> firstResponse = successResponses.values().iterator().next();
      callback.onSuccess(new ResponseImpl<>(firstResponse, gatheredResponse));
    }
    else
    {
      // all scattered requests are failing, we still return 200 for original request, but body will contain
      // failed response for each key.
      callback.onSuccess(new ResponseImpl<>(HttpStatus.S_200_OK.getCode(),
              Collections.emptyMap(), Collections.emptyList(), gatheredResponse, null));
    }
  }
@Test(dataProvider="illegalRequestMethodProvider", expectedExceptions = {UnsupportedOperationException.class})
  public void testUnsupportedRequestOnCompletion(ResourceMethod requestMethod)
  {
    Request<?> request = mock(Request.class);
    when(request.getMethod()).thenReturn(requestMethod);
    _sgStrategy.onAllResponsesReceived(request, AllProtocolVersions.LATEST_PROTOCOL_VERSION, Collections.emptyMap(),
            Collections.emptyMap(), Collections.emptyMap(), Callbacks.empty());
  }