import org.kitesdk.morphline.base.FaultTolerance; //导入依赖的package包/类
@Override
protected void setup(Context context) throws IOException, InterruptedException {
  verifyPartitionAssignment(context);    
  SolrRecordWriter.addReducerContext(context);
  Class<? extends UpdateConflictResolver> resolverClass = context.getConfiguration().getClass(
      UPDATE_CONFLICT_RESOLVER, RetainMostRecentUpdateConflictResolver.class, UpdateConflictResolver.class);
  
  this.resolver = ReflectionUtils.newInstance(resolverClass, context.getConfiguration());
  /*
   * Note that ReflectionUtils.newInstance() above also implicitly calls
   * resolver.configure(context.getConfiguration()) if the resolver
   * implements org.apache.hadoop.conf.Configurable
   */

  this.exceptionHandler = new FaultTolerance(
      context.getConfiguration().getBoolean(FaultTolerance.IS_PRODUCTION_MODE, false), 
      context.getConfiguration().getBoolean(FaultTolerance.IS_IGNORING_RECOVERABLE_EXCEPTIONS, false),
      context.getConfiguration().get(FaultTolerance.RECOVERABLE_EXCEPTION_CLASSES, SolrServerException.class.getName()));
  
  this.heartBeater = new HeartBeater(context);
}
 

import org.kitesdk.morphline.base.FaultTolerance; //导入依赖的package包/类
@Override
public void configure(Context context) {
  if (context.getString(FaultTolerance.RECOVERABLE_EXCEPTION_CLASSES) == null) {
    context.put(FaultTolerance.RECOVERABLE_EXCEPTION_CLASSES, 
        "org.apache.solr.client.solrj.SolrServerException");      
  }
  super.configure(context);
}
 

import org.kitesdk.morphline.base.FaultTolerance; //导入依赖的package包/类
@Override
public void configure(Context context) {
  String morphlineFile = context.getString(MORPHLINE_FILE_PARAM);
  String morphlineId = context.getString(MORPHLINE_ID_PARAM);
  if (morphlineFile == null || morphlineFile.trim().length() == 0) {
    throw new MorphlineCompilationException("Missing parameter: " + MORPHLINE_FILE_PARAM, null);
  }
  morphlineFileAndId = morphlineFile + "@" + morphlineId;
  
  if (morphlineContext == null) {
    FaultTolerance faultTolerance = new FaultTolerance(
        context.getBoolean(FaultTolerance.IS_PRODUCTION_MODE, false), 
        context.getBoolean(FaultTolerance.IS_IGNORING_RECOVERABLE_EXCEPTIONS, false),
        context.getString(FaultTolerance.RECOVERABLE_EXCEPTION_CLASSES));
    
    morphlineContext = new MorphlineContext.Builder()
      .setExceptionHandler(faultTolerance)
      .setMetricRegistry(SharedMetricRegistries.getOrCreate(morphlineFileAndId))
      .build();
  }
  
  Config override = ConfigFactory.parseMap(
      context.getSubProperties(MORPHLINE_VARIABLE_PARAM + "."));
  morphline = new Compiler().compile(
      new File(morphlineFile), morphlineId, morphlineContext, finalChild, override);
  
  this.mappingTimer = morphlineContext.getMetricRegistry().timer(
      MetricRegistry.name("morphline.app", Metrics.ELAPSED_TIME));
  this.numRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", Metrics.NUM_RECORDS));
  this.numFailedRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", "numFailedRecords"));
  this.numExceptionRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", "numExceptionRecords"));
}
 

import org.kitesdk.morphline.base.FaultTolerance; //导入依赖的package包/类
/**
 *
 * @param morphlineFile
 * @param morphlineId
 * @param collector
 * @param isProduction
 * @return
 */
public static Pipeline setPipeline(String morphlineFile, String morphlineId, Collector collector, boolean isProduction) {
  LOG.debug("Constructing Pipeline[{}#{}]", morphlineFile, morphlineId);

  // Set up the Morphline context and handler
  MorphlineContext context = new MorphlineContext.Builder()
      .setExceptionHandler(new FaultTolerance(isProduction, false))
      .build();

  // Compile the Morphline process
  Command morphline;
  try {
    morphline = new Compiler().compile(
        new File(morphlineFile),
        morphlineId,
        context,
        collector);
  } catch (Exception e) {
    throw new MorphlineCompilationException("Morphline compilation error", null, e);
  }

  // Create the pipeline wrapper
  Pipeline pipeline = new Pipeline(morphline, collector);

  // Ensure shutdown notification to Morphline commands esp in streaming environments
  JVMUtils.closeAtShutdown(pipeline);

  // Prep the pipeline
  Notifications.notifyBeginTransaction(pipeline.getMorphline());

  // Register the pipeline into the cache
  if (null == pipelineCache.get()) {
    pipelineCache.set(new HashMap<String, Pipeline>());
  }
  pipelineCache.get().put(morphlineFile + SEPARATOR + morphlineId, pipeline);

  LOG.trace("Pipeline[{}#{}] prepared", morphlineFile, morphlineId);
  return pipeline;
}
 

import org.kitesdk.morphline.base.FaultTolerance; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
@Override
public void prepare(Map stormConf, TopologyContext topologyContext, OutputCollector collector) {
    LOG.info("START prepare");

    this.collector = collector;

    File confFile = loadFile(topologyContext.getThisWorkerPort().toString());

    if (morphlineContext == null) {
        //TODO Make FaultTolerance configurable
        FaultTolerance faultTolerance = new FaultTolerance(true, false, null);

        morphlineContext = new MorphlineContext.Builder()
                .setExceptionHandler(faultTolerance)
                //.setMetricRegistry(SharedMetricRegistries.getOrCreate(customMorphlineId))
                .build();
    }

    Config override = ConfigFactory.parseMap(new HashMap<String, Object>());
    finalChild = new SimpleCommandCollector();
    morphline = new Compiler().compile(confFile, morphlineId, morphlineContext, finalChild, override);

    if (!EmptyUtils.nullOrEmpty(outputFieldNames)) {
        terminalBolt = false;
    }
}
 

import org.kitesdk.morphline.base.FaultTolerance; //导入依赖的package包/类
private MorphlineContext createMorphlineContext() {
    return new SolrMorphlineContext.Builder()
      .setDocumentLoader(testServer)
//      .setDocumentLoader(new CollectingDocumentLoader(100))
      .setExceptionHandler(new FaultTolerance(false, false, SolrServerException.class.getName()))
      .setMetricRegistry(new MetricRegistry())
      .build();
  }
 

import org.kitesdk.morphline.base.FaultTolerance; //导入依赖的package包/类
private MorphlineContext createMorphlineContext() {
  return new MorphlineContext.Builder()
    .setExceptionHandler(new FaultTolerance(false, false, SolrServerException.class.getName()))
    .setMetricRegistry(new MetricRegistry())
    .build();
}