source = context.newPluginInstance(sourcePluginId);
RealtimeContext sourceContext = new WorkerRealtimeContext(context, metrics, sourcePluginId);
LOG.debug("Source Stage : {}", stage.getName());
LOG.debug("Source Class : {}", source.getClass().getName());
source.initialize(sourceContext);
<BUG>sourceEmitter = new DefaultEmitter(new StageMetrics(metrics, StageMetrics.Type.SOURCE, stage.getName()));
}</BUG>
@SuppressWarnings("unchecked")
private void initializeSink(WorkerContext context, ETLStage stage) throws Exception {
String sinkPluginId = context.getRuntimeArguments().get(Constants.Sink.PLUGINID);

source = context.newPluginInstance(sourcePluginId);
RealtimeContext sourceContext = new WorkerRealtimeContext(context, metrics, sourcePluginId);
LOG.debug("Source Stage : {}", stage.getName());
LOG.debug("Source Class : {}", source.getClass().getName());
source.initialize(sourceContext);
sourceEmitter = new DefaultEmitter(new StageMetrics(metrics, PluginID.from(sourcePluginId)));
}
@SuppressWarnings("unchecked")
private void initializeSink(WorkerContext context, ETLStage stage) throws Exception {
String sinkPluginId = context.getRuntimeArguments().get(Constants.Sink.PLUGINID);
