@Test
public void testPutMetricsUsingOffsetIndex() throws IOException {

	Long currentTime = System.currentTimeMillis();
	ElasticSearchConsumerOffsetMetricsService spyService = _initializeSpyService(esConsumerOffsetMetricsService, successReply, true);
	List<Metric> metrics = new ArrayList<>();
	Metric record1 = new Metric("scope", "metric");
	record1.setTag("topic", "topicV");
	record1.addDatapoint(currentTime, 0.0);
	metrics.add(record1);
	Metric record2 = new Metric("scope2", "metric2");
	record2.setTag("topic", "topicV");
	record2.setTag("groupId", "groupId");
	record2.addDatapoint(currentTime, 1.0);
	metrics.add(record2);
	spyService.putMetrics(metrics);
	ArgumentCaptor<String> requestCaptorUrl = ArgumentCaptor.forClass(String.class);
	ArgumentCaptor<String> requestCaptorBody = ArgumentCaptor.forClass(String.class);
	verifyStatic(ElasticSearchUtils.class, times(1));
	ElasticSearchUtils.performESRequest(eq(restClient), requestCaptorUrl.capture(), requestCaptorBody.capture());
	String expectedURL = "_bulk";
	assertEquals(expectedURL, requestCaptorUrl.getValue());
	String[] lines = requestCaptorBody.getValue().split("\\r?\\n");
	String expectedIndexName = getExpectedIndexName();
	JsonNode root = mapper.readTree(lines[0]);
	String actualIndexName = root.get("index").get("_index").asText();
	assertEquals(expectedIndexName, actualIndexName);
	String expectedSerializedMetric1 =
			"{\"metric\":\"metric\",\"topic\":\"topicV\",\"value\":\"0.0\",\"ts\":\"" + currentTime + "\",\"tags\":\"{\\\"service\\\":\\\"scope\\\"}\"}";
	String expectedSerializedMetric2 =
			"{\"metric\":\"metric2\",\"topic\":\"topicV\",\"value\":\"1.0\",\"ts\":\"" + currentTime + "\",\"tags\":\"{\\\"groupId\\\":\\\"groupId\\\",\\\"service\\\":\\\"scope2\\\"}\"}";
	List<String> expectedOutput = new ArrayList<>(Arrays.asList(expectedSerializedMetric1, expectedSerializedMetric2));
	Collections.sort(expectedOutput);
	List<String> actualOutput = new ArrayList<>(Arrays.asList(lines[1], lines[3]));
	Collections.sort(actualOutput);
	assertEquals(Arrays.asList(expectedOutput), Arrays.asList(actualOutput));
}