class AsyncCallableService implements Callable<T> {

  @Override
  public T call() throws Exception {
    long startTime = System.currentTimeMillis();
    long timeLeft = timeout;
    Future<T> future = executorService.submit(task);
    LOG.info("Task {} execution started at {}", taskName, startTime);

    Throwable lastError = null;
    while (true) {
      try {
        LOG.debug("Task {} waiting for result at most {} ms", taskName, timeLeft);
        T taskResult = future.get(timeLeft, TimeUnit.MILLISECONDS);
        LOG.info("Task {} successfully completed with result: {}", taskName, taskResult);
        return taskResult;
      } catch (TimeoutException e) {
        LOG.debug("Task {} timeout", taskName);
        if (lastError == null) {
          lastError = e;
        }
        timeLeft = 0;
      } catch (ExecutionException e) {
        Throwable cause = Throwables.getRootCause(e);
        if (!(cause instanceof RetryTaskSilently)) {
          LOG.info(String.format("Task %s exception during execution", taskName), cause);
        }
        lastError = cause;
        timeLeft = timeout - (System.currentTimeMillis() - startTime) - retryDelay;
      }

      if (timeLeft <= 0) {
        attemptToCancel(future);
        LOG.warn("Task {} timeout exceeded, no more retries", taskName);
        onError.accept(lastError);
        return null;
      }

      LOG.debug("Task {} retrying execution in {} milliseconds", taskName, retryDelay);
      future = executorService.schedule(task, retryDelay, TimeUnit.MILLISECONDS);
    }
  }

  public  AsyncCallableService(Callable<T> task, long timeout, long retryDelay, String taskName, Consumer<Throwable> onError);
  public  AsyncCallableService(Callable<T> task, long timeout, long retryDelay, String taskName, ScheduledExecutorService executorService, Consumer<Throwable> onError);

}