@Test
@PrepareForTest({URL.class, OutputStream.class, AbstractTimelineMetricsSink.class, HttpURLConnection.class})

public void testPutMetrics() throws Exception {
  HadoopTimelineMetricsSink sink = new HadoopTimelineMetricsSink();
  HttpURLConnection connection = PowerMock.createNiceMock(HttpURLConnection.class);
  URL url = PowerMock.createNiceMock(URL.class);
  InputStream is = IOUtils.toInputStream(gson.toJson(Collections.singletonList("localhost")));
  expectNew(URL.class, anyString()).andReturn(url).anyTimes();
  expect(url.openConnection()).andReturn(connection).anyTimes();
  expect(connection.getInputStream()).andReturn(is).anyTimes();
  expect(connection.getResponseCode()).andReturn(200).anyTimes();
  OutputStream os = PowerMock.createNiceMock(OutputStream.class);
  expect(connection.getOutputStream()).andReturn(os).anyTimes();
  SubsetConfiguration conf = createNiceMock(SubsetConfiguration.class);
  expect(conf.getString("slave.host.name")).andReturn("localhost").anyTimes();
  expect(conf.getParent()).andReturn(null).anyTimes();
  expect(conf.getPrefix()).andReturn("service").anyTimes();
  expect(conf.getStringArray(eq(COLLECTOR_HOSTS_PROPERTY))).andReturn(new String[]{"localhost"," localhost2"}).anyTimes();
  expect(conf.getString(eq("serviceName-prefix"), eq(""))).andReturn("").anyTimes();
  expect(conf.getString(eq(COLLECTOR_PROTOCOL), eq("http"))).andReturn("http").anyTimes();
  expect(conf.getString(eq(COLLECTOR_PORT), eq("6188"))).andReturn("6188").anyTimes();
  expect(conf.getInt(eq(MAX_METRIC_ROW_CACHE_SIZE), anyInt())).andReturn(10).anyTimes();
  expect(conf.getInt(eq(METRICS_SEND_INTERVAL), anyInt())).andReturn(1000).anyTimes();
  conf.setListDelimiter(eq(','));
  expectLastCall().anyTimes();
  expect(conf.getKeys()).andReturn(new Iterator() {
    @Override
    public boolean hasNext() {
      return false;
    }
    @Override
    public Object next() {
      return null;
    }
    @Override
    public void remove() {
    }
  }).once();
  AbstractMetric metric = createNiceMock(AbstractMetric.class);
  expect(metric.name()).andReturn("metricName").anyTimes();
  expect(metric.value()).andReturn(9.5687).anyTimes();
  expect(metric.type()).andReturn(MetricType.COUNTER).anyTimes();
  //TODO currently only numeric metrics are supported
  MetricsRecord record = createNiceMock(MetricsRecord.class);
  expect(record.name()).andReturn("testName").anyTimes();
  expect(record.context()).andReturn("testContext").anyTimes();
  expect(record.timestamp()).andAnswer(new IAnswer<Long>() {
    @Override
    public Long answer() throws Throwable {
      return System.currentTimeMillis();
    }
  }).anyTimes();
  expect(record.metrics()).andReturn(Arrays.asList(metric)).anyTimes();
  replay(conf, record, metric);
  replayAll();
  sink.init(conf);
  sink.putMetrics(record);
  Thread.sleep(1500L);
  sink.putMetrics(record);
  verifyAll();
}