public void initialize(MapReduceContext context) throws Exception {
context.getSpecification().getProperties();
Map<String, String> properties = context.getSpecification().getProperties();
String sourcePluginId = properties.get(Constants.Source.PLUGINID);
String sinkPluginId = properties.get(Constants.Sink.PLUGINID);
<BUG>List<String> transformIds = GSON.fromJson(properties.get(Constants.Transform.PLUGINIDS), STRING_LIST_TYPE);

transforms = Lists.newArrayListWithCapacity(transformIds.size());
List<Transformation> pipeline = Lists.newArrayListWithCapacity(transformIds.size() + 2);

List<StageMetrics> stageMetrics = Lists.newArrayListWithCapacity(pipeline.size());</BUG>
BatchSource source = context.newInstance(sourcePluginId);

public void initialize(MapReduceContext context) throws Exception {
context.getSpecification().getProperties();
Map<String, String> properties = context.getSpecification().getProperties();
String sourcePluginId = properties.get(Constants.Source.PLUGINID);
String sinkPluginId = properties.get(Constants.Sink.PLUGINID);
List<TransformInfo> transformInfos = GSON.fromJson(properties.get(Constants.Transform.PLUGINIDS),
TRANSFORMDETAILS_LIST_TYPE);
List<TransformDetail> pipeline = Lists.newArrayListWithCapacity(transformInfos.size() + 2);
BatchSource source = context.newInstance(sourcePluginId);
