@Override
  public <T> DistributedAnswer<T> call(DistributedCall<T> callable, MemberSelector memberSelector, long timeoutMs)
    throws InterruptedException {

    IExecutorService executor = hzInstance.getExecutorService("default");
    Map<Member, Future<T>> futures = executor.submitToMembers(callable, memberSelector);
    try {
      DistributedAnswer<T> distributedAnswer = new DistributedAnswer<>();
      long maxTime = System.currentTimeMillis() + timeoutMs;
      for (Map.Entry<Member, Future<T>> entry : futures.entrySet()) {
        long remainingMs = Math.max(maxTime - System.currentTimeMillis(), 5L);
        try {
          T answer = entry.getValue().get(remainingMs, TimeUnit.MILLISECONDS);
          distributedAnswer.setAnswer(entry.getKey(), answer);
        } catch (ExecutionException e) {
          distributedAnswer.setFailed(entry.getKey(), e);
        } catch (TimeoutException e) {
          distributedAnswer.setTimedOut(entry.getKey());
        }
      }
      return distributedAnswer;
    } finally {
      futures.values().forEach(f -> f.cancel(true));
    }
  }
@Test
  public void timed_out_calls_do_not_break_other_answers() throws InterruptedException {
    // member 1 and 3 success, member 2 times-out
    TimedOutDistributedCall.COUNTER.set(0L);
    DistributedCall call = new TimedOutDistributedCall();
    DistributedAnswer<Long> answer = member1.call(call, MemberSelectors.DATA_MEMBER_SELECTOR, 2_000L);

    assertThat(extractAnswers(answer)).containsOnlyOnce(0L, 2L);

    assertThat(extractTimeOuts(answer)).containsExactlyInAnyOrder(false, false, true);
  }