@Override
  protected OptimizingTaskResult executeTask(OptimizingTask task) {
    OptimizingTaskResult result = executeTask(getConfig(), getThreadId(), task, LOG);
    callBeforeTaskComplete();
    // add optimizer flink runtime info, including application_id, tm_id
    StringBuilder sb = new StringBuilder();
    if (!Strings.isNullOrEmpty(result.getErrorMessage())) {
      if (runtimeContext != null && !runtimeContext.isEmpty()) {
        runtimeContext.forEach((k, v) -> sb.append(k).append("=").append(v).append("\n"));
      }
      String errorMsg = sb + result.getErrorMessage();
      result.setErrorMessage(
          errorMsg.substring(0, Math.min(ERROR_MESSAGE_MAX_LENGTH, errorMsg.length())));
    }
    return result;
  }
Optimizer executor For Flink engine supports:

<ul>
  <li>Add additional content to error messages to help locate execution nodes.
  <li>Add some Flink metrics like task number executed.
</ul>
amoro-optimizer/amoro-optimizer-flink/src/main/java/org/apache/amoro/optimizer/flink/FlinkOptimizerExecutor.java