code: @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();
          }
      }
docs: 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.
  
  If we miss the offset, we'll sleep until the offset comes next minute.