@Override
  public void writeMetrics(MetricQueryResults metricQueryResults) throws Exception {
    URL url = new URL(urlString);
    String metrics = serializeMetrics(metricQueryResults);
    byte[] postData = metrics.getBytes(StandardCharsets.UTF_8);
    HttpURLConnection connection = (HttpURLConnection) url.openConnection();
    connection.setDoOutput(true);
    connection.setInstanceFollowRedirects(false);
    connection.setRequestMethod("POST");
    connection.setRequestProperty("Content-Type", "application/json");
    connection.setRequestProperty("charset", "utf-8");
    connection.setRequestProperty("Content-Length", Integer.toString(postData.length));
    connection.setUseCaches(false);
    try (DataOutputStream connectionOuputStream =
        new DataOutputStream(connection.getOutputStream())) {
      connectionOuputStream.write(postData);
    }
    int responseCode = connection.getResponseCode();
    if (responseCode != 200) {
      throw new IOException(
          "Expected HTTP 200 OK response while writing metrics to MetricsHttpSink but received "
              + responseCode);
    }
  }
@Test
  public void testWriteMetricsWithCommittedSupported() throws Exception {
    MetricQueryResults metricQueryResults = new CustomMetricQueryResults(true);
    MetricsOptions pipelineOptions = PipelineOptionsFactory.create().as(MetricsOptions.class);
    pipelineOptions.setMetricsHttpSinkUrl(String.format("http://localhost:%s", port));
    MetricsHttpSink metricsHttpSink = new MetricsHttpSink(pipelineOptions);
    countDownLatch = new CountDownLatch(1);
    metricsHttpSink.writeMetrics(metricQueryResults);
    countDownLatch.await();
    String expected =
        "{\"counters\":[{\"attempted\":20,\"committed\":10,\"name\":{\"name\":\"n1\","
            + "\"namespace\":\"ns1\"},\"step\":\"s1\"}],\"distributions\":[{\"attempted\":"
            + "{\"count\":4,\"max\":9,\"mean\":6.25,\"min\":3,\"sum\":25},\"committed\":"
            + "{\"count\":2,\"max\":8,\"mean\":5.0,\"min\":5,\"sum\":10},\"name\":{\"name\":\"n2\","
            + "\"namespace\":\"ns1\"},\"step\":\"s2\"}],\"gauges\":[{\"attempted\":{\"timestamp\":"
            + "\"1970-01-05T00:04:22.800Z\",\"value\":120},\"committed\":{\"timestamp\":"
            + "\"1970-01-05T00:04:22.800Z\",\"value\":100},\"name\":{\"name\":\"n3\",\"namespace\":"
            + "\"ns1\"},\"step\":\"s3\"}],\"stringSets\":[{\"attempted\":{\"stringSet\":[\"cd"
            + "\"]},\"committed\":{\"stringSet\":[\"ab\"]},\"name\":{\"name\":\"n3\","
            + "\"namespace\":\"ns1\"},\"step\":\"s3\"}]}";
    assertEquals("Wrong number of messages sent to HTTP server", 1, messages.size());
    assertEquals("Wrong messages sent to HTTP server", expected, messages.get(0));
  }