guavasupport/src/main/java/com/streamsets/pipeline/lib/io/fileref/FileRefUtil.java
@SuppressWarnings("unchecked")
  public static synchronized void initMetricsIfNeeded(ProtoConfigurableEntity.Context context) {
    Gauge<Map<String, Object>> gauge = context.getGauge(fileStatisticGaugeName(context));
    if(gauge == null) {
      gauge = context.createGauge(fileStatisticGaugeName(context), Comparator.comparing(GAUGE_MAP_ORDERING::get));
      Map<String, Object> gaugeStatistics = gauge.getValue();
      //File name is populated at the MetricEnabledWrapperStream.
      gaugeStatistics.put(FileRefUtil.FILE, "");
      gaugeStatistics.put(FileRefUtil.TRANSFER_THROUGHPUT, 0L);
      gaugeStatistics.put(FileRefUtil.SENT_BYTES, String.format(FileRefUtil.BRACKETED_TEMPLATE, 0, 0));
      gaugeStatistics.put(FileRefUtil.REMAINING_BYTES, 0L);
      gaugeStatistics.put(FileRefUtil.COMPLETED_FILE_COUNT, 0L);
    }

    Meter dataTransferMeter = context.getMeter(FileRefUtil.TRANSFER_THROUGHPUT_METER);
    if (dataTransferMeter == null) {
      context.createMeter(FileRefUtil.TRANSFER_THROUGHPUT_METER);
    }
  }
@SuppressWarnings("unchecked")
  public static synchronized void initMetricsIfNeeded(ProtoConfigurableEntity.Context context) {
    Gauge<Map<String, Object>> gauge = context.getGauge(fileStatisticGaugeName(context));
    if(gauge == null) {
      gauge = context.createGauge(fileStatisticGaugeName(context), Comparator.comparing(GAUGE_MAP_ORDERING::get));
      Map<String, Object> gaugeStatistics = gauge.getValue();
      //File name is populated at the MetricEnabledWrapperStream.
      gaugeStatistics.put(FileRefUtil.FILE, "");
      gaugeStatistics.put(FileRefUtil.TRANSFER_THROUGHPUT, 0L);
      gaugeStatistics.put(FileRefUtil.SENT_BYTES, String.format(FileRefUtil.BRACKETED_TEMPLATE, 0, 0));
      gaugeStatistics.put(FileRefUtil.REMAINING_BYTES, 0L);
      gaugeStatistics.put(FileRefUtil.COMPLETED_FILE_COUNT, 0L);
    }

    Meter dataTransferMeter = context.getMeter(FileRefUtil.TRANSFER_THROUGHPUT_METER);
    if (dataTransferMeter == null) {
      context.createMeter(FileRefUtil.TRANSFER_THROUGHPUT_METER);
    }
  }
Creates a gauge if it is already not. This is done only once for the stage
@param context the {@link com.streamsets.pipeline.api.Stage.Context} of this stage