@Override
  @SuppressWarnings("unchecked")
  public <T extends Serializable> T executeOrThrowException(
      String idempotentId, String itemName, Callable<T> callable) throws Exception {
    String jobIdPrefix = "Job " + jobId + ": ";

    RetryingCallable<T> retryingCallable =
        new RetryingCallable<>(
            callable,
            retryStrategyLibrary,
            Clock.systemUTC(),
            monitor);

    if (knownValues.containsKey(idempotentId)) {
      monitor.debug(
          () ->
              jobIdPrefix
                  + format("Using cached key %s from cache for %s", idempotentId, itemName));
      return (T) knownValues.get(idempotentId);
    }
    try {
      T result = retryingCallable.call();
      knownValues.put(idempotentId, result);
      monitor.debug(
          () -> jobIdPrefix + format("Storing key %s in cache for %s", idempotentId, itemName));
      errors.remove(idempotentId);
      return result;
    } catch (RetryException e) {
      ErrorDetail.Builder errorDetailBuilder = ErrorDetail.builder();
      errorDetailBuilder.setId(idempotentId)
          .setTitle(itemName)
          .setException(Throwables.getStackTraceAsString(e));
      if(e.canSkip()){
        ErrorDetail errorDetail = errorDetailBuilder.setCanSkip(true).build();
        errors.put(idempotentId, errorDetail);
        recentErrors.put(idempotentId, errorDetail);
        monitor.severe(() -> jobIdPrefix + "Problem with importing item, but skipping: " + errorDetail);
        return  null;
      } else {
        ErrorDetail errorDetail = errorDetailBuilder.build();
        errors.put(idempotentId, errorDetail);
        recentErrors.put(idempotentId, errorDetail);
        monitor.severe(() -> jobIdPrefix + "Problem with importing item, cannot be skipped: " + errorDetail);
        // In RetryingCallable.call(), we wrap the exception in a RetryException.
        // We throw e.getCause() to get the original exception.
        throw e.getCause();
      }
    }
  }
A {@link IdempotentImportExecutor} that stores known values in memory.
portability-spi-transfer/src/main/java/org/datatransferproject/spi/transfer/idempotentexecutor/RetryingInMemoryIdempotentImportExecutor.java