public void reset() {
    triggerFuture = new CompletableFuture<>();
    callbackFuture = triggerFuture;
    callbackMap.keySet().forEach(this::whenCompleted);
  }
Only proceed the callback if there is no exception. Exceptions would be thrown in the
complete() method and will not trigger join return.
amoro-common/src/main/java/org/apache/amoro/process/SimpleFuture.java