private List<StageOutput> addReportedErrorsIfNeeded(List<StageOutput> snapshotsOfAllStagesOutput) {
    synchronized (this.reportedErrors) {
      if(reportedErrors.isEmpty()) {
        return snapshotsOfAllStagesOutput;
      }

      try {
        return snapshotsOfAllStagesOutput.stream()
          .map(so -> new StageOutput(
            so.getInstanceName(),
            so.getOutput(),
            so.getErrorRecords(),
            reportedErrors.get(so.getInstanceName()),
            so.getEventRecords()
          ))
          .collect(Collectors.toList());
      } finally {
        reportedErrors.clear();
      }
    }
  }
private List < StageOutput > addReportedErrorsIfNeeded ( List < StageOutput > snapshotsOfAllStagesOutput ) { synchronized ( this . reportedErrors ) { if ( reportedErrors . isEmpty ( ) ) { return snapshotsOfAllStagesOutput ; } try { return snapshotsOfAllStagesOutput . stream ( ) . map ( so -> new StageOutput ( so . getInstanceName ( ) , so . getOutput ( ) , so . getErrorRecords ( ) , reportedErrors . get ( so . getInstanceName ( ) ) , so . getEventRecords ( ) ) ) . collect ( Collectors . toList ( ) ) ; } finally { reportedErrors . clear ( ) ; } } }
Preview only returns data associated with batches, however errors are reported outside of batch context for
multi-threaded pipelines. Thus we 'emulate' the behavior by simply adding into the current batch all 'so-far'
reported errors.