@ Override public void process ( Object event ) throws Exception { int secOffset = TimeUtils . secOffset ( ) ; if ( secOffset < UPLOAD_TIME_OFFSET_SEC ) { JStormUtils . sleepMs ( ( UPLOAD_TIME_OFFSET_SEC - secOffset ) * 1000 ) ; } else if ( secOffset == UPLOAD_TIME_OFFSET_SEC ) { // do nothing } else { JStormUtils . sleepMs ( ( 60 - secOffset + UPLOAD_TIME_OFFSET_SEC ) * 1000 ) ; } if ( topologyMetricContext . getUploadedWorkerNum ( ) > 0 ) { metricLogger . info ( "force upload metrics." ) ; mergeAndUploadMetrics ( ) ; } }
Wait UPLOAD_TIME_OFFSET_SEC sec to ensure we ve collected enough metrics from topology workers note that it s not guaranteed metrics from all workers will be collected .