MetricReporter metricReporter = workerData.getMetricsReporter();
boolean isMetricsEnable = ConfigExtension
.isEnablePerformanceMetrics(workerData.getStormConf());
metricReporter.setEnable(isMetricsEnable);
metricReporter.start();
LOG.info("Start metrics reporter, enable performance metrics: "
+ isMetricsEnable);
// create task heartbeat
TaskHeartbeatRunable taskHB = new TaskHeartbeatRunable(workerData);