java.util.concurrent.ExecutionException#toString ( )源码实例Demo

@Override
public SnapshotInfo save(
    String name,
    String rev,
    String id,
    long batchNumber,
    List<List<StageOutput>> snapshotBatches
) throws PipelineException {
  synchronized (lockCache.getLock(name)) {
    try {
      SnapshotInfo snapshotInfo = getSnapshotInfoFromCache(name, rev, id);
      if (snapshotInfo == null) {
        throw new PipelineException(ContainerError.CONTAINER_0605);
      }
      SnapshotInfo updatedSnapshotInfo = snapshotStore.save(name, rev, id, batchNumber, snapshotBatches);
      snapshotStateCache.put(getCacheKey(name, rev, id), updatedSnapshotInfo);
      return updatedSnapshotInfo;
    } catch (ExecutionException e) {
      throw new PipelineException(ContainerError.CONTAINER_0600, id, name, rev, e.toString(), e);
    }
  }
}
 

/**
 * Waits until the result of {@link CompletableFuture} which is completed after the {@link #close()} or
 * {@link #closeAsync()} operation is completed.
 */
public void blockUntilShutdown() throws InterruptedException {
    try {
        whenClosed().get();
    } catch (ExecutionException e) {
        throw new CompletionException(e.toString(), Exceptions.peel(e));
    }
}
 

@Override
public Path getPath(FileSystem fs, Date recordDate, Record record) throws StageException, IOException {
  // runUuid is fixed for the current pipeline run. it avoids collisions with other SDCs running the same/similar
  // pipeline
  try {
    return dirPathCache.get(recordWriterManager.getDirPath(recordDate, record));
  } catch (ExecutionException ex) {
    if (ex.getCause() instanceof StageException) {
      throw (StageException) ex.getCause();
    } else{
      throw new StageException(Errors.HADOOPFS_24, ex.toString(), ex);
    }
  }
}
 

@Override
public PipelineState getState(String name, String rev) throws PipelineStoreException  {
  try {
    return pipelineStateCache.get(getNameAndRevString(name, rev));
  } catch (ExecutionException ex) {
    if (ex.getCause() instanceof RuntimeException) {
      throw (RuntimeException) ex.getCause();
    } else if (ex.getCause() instanceof PipelineStoreException) {
      throw (PipelineStoreException) ex.getCause();
    } else {
      throw new PipelineStoreException(ContainerError.CONTAINER_0114, ex.toString(), ex);
    }
  }
}
 

@Override
@SuppressWarnings("deprecation")
public Runner getRunner(final String name, final String rev) throws PipelineException {
  if (!pipelineStore.hasPipeline(name)) {
    throw new PipelineStoreException(ContainerError.CONTAINER_0200, name);
  }
  final String nameAndRevString = getNameAndRevString(name, rev);
  RunnerInfo runnerInfo;
  try {
    runnerInfo = runnerCache.get(nameAndRevString, () -> {
      ExecutionMode executionMode = pipelineStateStore.getState(name, rev).getExecutionMode();
      Runner runner = getRunner(name, rev, executionMode);
      return new RunnerInfo(runner, executionMode);
    });
    ExecutionMode cachedExecutionMode = runnerInfo.executionMode;
    ExecutionMode persistentExecutionMode = pipelineStateStore.getState(name, rev).getExecutionMode();
    if (cachedExecutionMode == ExecutionMode.CLUSTER) {
      LOG.info("Upgrading execution mode from " + ExecutionMode.CLUSTER + " to " + persistentExecutionMode);
      runnerInfo.executionMode = persistentExecutionMode;
    }
    if (runnerInfo.executionMode != pipelineStateStore.getState(name, rev).getExecutionMode()) {
      LOG.info(Utils.format("Invalidate the existing runner for pipeline '{}::{}' as execution mode has changed",
        name, rev));
      if (!removeRunnerIfNotActive(runnerInfo.runner)) {
        throw new PipelineManagerException(ValidationError.VALIDATION_0082, pipelineStateStore.getState(name, rev).getExecutionMode(),
          runnerInfo.executionMode);
      } else {
        return getRunner(name, rev);
      }
    }
  } catch (ExecutionException ex) {
    if (ex.getCause() instanceof RuntimeException) {
      throw (RuntimeException) ex.getCause();
    } else if (ex.getCause() instanceof PipelineStoreException) {
      throw (PipelineStoreException) ex.getCause();
    } else {
      throw new PipelineStoreException(ContainerError.CONTAINER_0114, ex.toString(), ex);
    }
  }
  return runnerInfo.runner;
}
 

@Override
protected void process(Record record, SingleLaneBatchMaker batchMaker) throws StageException {


  Map<String, String> fromFieldToFieldMap;
  Set<String> fieldsRequiringOverwrite;
  Set<String> fieldsThatDoNotExist;
  Map<String, Set<String>> multipleRegexMatchingSameFields;

  CachedResults data;
  try {
    data = cache.get(record.getEscapedFieldPaths());
    count++;
    if(count == 50000) {
      count = 0;
      CacheStats stats = cache.stats();
      LOG.debug("cache_stats: hits {} misses {} rate {} ",
          stats.hitCount(),
          stats.missCount(),
          stats.hitRate());
    }
  } catch (ExecutionException ex) {
    LOG.error(Errors.FIELD_RENAMER_05.getMessage(), ex.toString(), ex);
    throw new StageException(Errors.FIELD_RENAMER_05, ex.toString(), ex);
  }

  fromFieldToFieldMap = data.fromFieldToFieldMap;
  fieldsRequiringOverwrite = data.fieldsRequiringOverwrite;
  fieldsThatDoNotExist = data.fieldsThatDoNotExist;
  multipleRegexMatchingSameFields = data.multipleRegexMatchingSameFields;

  for (Map.Entry<String, String> fromFieldToFieldEntry : fromFieldToFieldMap.entrySet()) {
    String fromFieldName = fromFieldToFieldEntry.getKey();
    String toFieldName = fromFieldToFieldEntry.getValue();
    // The fromFieldName will always exist, not need to check for its existence.
    // If the source field exists and the target does not, we need to replace
    // We can also replace in this case if overwrite existing is set to true
    Field fromField = record.get(fromFieldName);
    if (record.has(toFieldName)) {
      switch (errorHandler.existingToFieldHandling) {
        case TO_ERROR:
          fieldsRequiringOverwrite.add(toFieldName);
          break;
        case CONTINUE:
          break;
        case APPEND_NUMBERS:
          int i = 1;
          for (; record.has(toFieldName + i); i++);
          toFieldName = toFieldName + i;
          //Fall through so as to edit.
        case REPLACE:
          // No need to bother if the field is being renamed to itself; plus, we don't want to delete it either
          if (!toFieldName.equals(fromFieldName)) {
            record.set(toFieldName, fromField);
            record.delete(fromFieldName);
          }
          break;
      }
    } else {
      try {
        record.set(toFieldName, fromField);
      } catch (IllegalArgumentException e) {
        throw new OnRecordErrorException(record, Errors.FIELD_RENAMER_04, toFieldName, e.toString());
      }
      record.delete(fromFieldName);
    }
  }

  if (errorHandler.nonExistingFromFieldHandling == OnStagePreConditionFailure.TO_ERROR
      && !fieldsThatDoNotExist.isEmpty()) {
    throw new OnRecordErrorException(Errors.FIELD_RENAMER_00, record.getHeader().getSourceId(),
        Joiner.on(", ").join(fieldsThatDoNotExist));
  }

  if (errorHandler.multipleFromFieldsMatching == OnStagePreConditionFailure.TO_ERROR
      && !multipleRegexMatchingSameFields.isEmpty()) {
    StringBuilder sb = new StringBuilder();
    for (Map.Entry<String, Set<String>> multipleRegexMatchingSameFieldEntry
        : multipleRegexMatchingSameFields.entrySet()) {
      sb.append(" Field: ")
          .append(multipleRegexMatchingSameFieldEntry.getKey())
          .append(" Regex: ")
          .append(Joiner.on(",").join(multipleRegexMatchingSameFieldEntry.getValue()));
    }
    throw new OnRecordErrorException(record, Errors.FIELD_RENAMER_03, sb.toString());
  }

  if (!fieldsRequiringOverwrite.isEmpty()) {
    throw new OnRecordErrorException(record, Errors.FIELD_RENAMER_01,
        Joiner.on(", ").join(fieldsRequiringOverwrite),
        record.getHeader().getSourceId());
  }

  batchMaker.addRecord(record);
}