@Override
  public void writeMetrics(MetricQueryResults metricQueryResults) throws Exception {
    final long metricTimestamp = System.currentTimeMillis() / 1000L;
    Socket socket = new Socket(InetAddress.getByName(address), port);
    BufferedWriter writer =
        new BufferedWriter(new OutputStreamWriter(socket.getOutputStream(), charset));
    StringBuilder messagePayload = new StringBuilder();
    Iterable<MetricResult<Long>> counters = metricQueryResults.getCounters();
    Iterable<MetricResult<GaugeResult>> gauges = metricQueryResults.getGauges();
    Iterable<MetricResult<DistributionResult>> distributions =
        metricQueryResults.getDistributions();

    for (MetricResult<Long> counter : counters) {
      messagePayload.append(new CounterMetricMessage(counter, "value", metricTimestamp).toString());
    }

    for (MetricResult<GaugeResult> gauge : gauges) {
      messagePayload.append(new GaugeMetricMessage(gauge, "value").toString());
    }

    for (MetricResult<DistributionResult> distribution : distributions) {
      messagePayload.append(
          new DistributionMetricMessage(distribution, "min", metricTimestamp).toString());
      messagePayload.append(
          new DistributionMetricMessage(distribution, "max", metricTimestamp).toString());
      messagePayload.append(
          new DistributionMetricMessage(distribution, "count", metricTimestamp).toString());
      messagePayload.append(
          new DistributionMetricMessage(distribution, "sum", metricTimestamp).toString());
      messagePayload.append(
          new DistributionMetricMessage(distribution, "mean", metricTimestamp).toString());
    }
    writer.write(messagePayload.toString());
    writer.flush();
    writer.close();
    socket.close();
  }
@Test
  public void testWriteMetricsWithCommittedUnSupported() throws Exception {
    MetricQueryResults metricQueryResults = new CustomMetricQueryResults(false);
    MetricsOptions pipelineOptions = PipelineOptionsFactory.create().as(MetricsOptions.class);
    pipelineOptions.setMetricsGraphitePort(port);
    pipelineOptions.setMetricsGraphiteHost("127.0.0.1");
    MetricsGraphiteSink metricsGraphiteSink = new MetricsGraphiteSink(pipelineOptions);
    CountDownLatch countDownLatch = new CountDownLatch(1);
    graphiteServer.setCountDownLatch(countDownLatch);
    metricsGraphiteSink.writeMetrics(metricQueryResults);
    countDownLatch.await();
    String join = String.join("\n", graphiteServer.getMessages());
    String regexpr =
        "beam.counter.ns1.n1.s1.attempted.value 20 [0-9]+\\n"
            + "beam.gauge.ns1.n3.s3.attempted.value 120 [0-9]+\\n"
            + "beam.distribution.ns1.n2.s2.attempted.min 3 [0-9]+\\n"
            + "beam.distribution.ns1.n2.s2.attempted.max 9 [0-9]+\\n"
            + "beam.distribution.ns1.n2.s2.attempted.count 4 [0-9]+\\n"
            + "beam.distribution.ns1.n2.s2.attempted.sum 25 [0-9]+\\n"
            + "beam.distribution.ns1.n2.s2.attempted.mean 6.25 [0-9]+";
    assertTrue(join.matches(regexpr));
  }