transforms = Lists.newArrayListWithCapacity(stageList.size());
BatchSource source = context.newPluginInstance(sourcePluginId);
BatchSourceContext batchSourceContext = new MapReduceSourceContext(context, mapperMetrics, sourcePluginId);
source.initialize(batchSourceContext);
pipeline.add(source);
<BUG>stageMetrics.add(new StageMetrics(mapperMetrics, StageMetrics.Type.SOURCE, etlConfig.getSource().getName()));
addTransforms(stageList, pipeline, stageMetrics, transformIds, context);</BUG>
BatchSink sink = context.newPluginInstance(sinkPluginId);
BatchSinkContext batchSinkContext = new MapReduceSinkContext(context, mapperMetrics, sinkPluginId);
sink.initialize(batchSinkContext);

transforms = Lists.newArrayListWithCapacity(stageList.size());
BatchSource source = context.newPluginInstance(sourcePluginId);
BatchSourceContext batchSourceContext = new MapReduceSourceContext(context, mapperMetrics, sourcePluginId);
source.initialize(batchSourceContext);
pipeline.add(source);
stageMetrics.add(new StageMetrics(mapperMetrics, PluginID.from(sourcePluginId)));
addTransforms(stageList, pipeline, stageMetrics, transformIds, context);
BatchSink sink = context.newPluginInstance(sinkPluginId);
BatchSinkContext batchSinkContext = new MapReduceSinkContext(context, mapperMetrics, sinkPluginId);
sink.initialize(batchSinkContext);
