ElasticSearchConsumerOffsetMetricsService extends DefaultService implements MetricStorageService { @VisibleForTesting static ObjectMapper getMetricObjectMapper(JsonSerializer<ConsumerOffsetRecordList> serializer, JsonDeserializer<ConsumerOffsetRecordList> deserializer) { ObjectMapper mapper = new ObjectMapper(); mapper.setSerializationInclusion(JsonInclude.Include.NON_NULL); SimpleModule module = new SimpleModule(); module.addSerializer(ConsumerOffsetRecordList.class, serializer); module.addDeserializer(ConsumerOffsetRecordList.class, deserializer); mapper.registerModule(module); return mapper; } @Inject protected  ElasticSearchConsumerOffsetMetricsService(SystemConfiguration config, MonitorService monitorService, ElasticSearchUtils esUtils); @Override void dispose(); @Override void putMetrics(List<Metric> metrics); @Override Map<MetricQuery, List<Metric>> getMetrics(List<MetricQuery> queries); static final int INDEX_MAX_RESULT_WINDOW; static final String DATE_FORMAT; static final String INDEX_FORMAT; static String INDEX_TEMPLATE_PATTERN_START; static final String SCOPE_NAME; static final String METRIC_NAME; static final String requestUrl; }
@Ignore @Test public void testConsumerOffsetSchemaRecordListMapper() throws IOException { mapper = ElasticSearchConsumerOffsetMetricsService.getMetricObjectMapper(new ConsumerOffsetRecordList.IndexSerializer(), new ConsumerOffsetRecordList.Deserializer()); Long cTime = System.currentTimeMillis(); ConsumerOffsetMetric record1 = new ConsumerOffsetMetric("metric", "topic", cTime, 0.0, defaultTags); ConsumerOffsetRecordList recordList = new ConsumerOffsetRecordList(Arrays.asList(record1), ElasticSearchUtils.HashAlgorithm.fromString("MD5")); String serialized = mapper.writeValueAsString(recordList); String[] lines = serialized.split("\\r?\\n"); String expectedIndexName = getExpectedIndexName(); JsonNode root = mapper.readTree(lines[0]); String actualIndexName = root.get("index").get("_index").asText(); assertEquals(expectedIndexName, actualIndexName); String expectedSerializedMetric = "{\"metric\":\"metric\",\"topic\":\"topic\",\"value\":\"0.0\",\"ts\":\"" + cTime + "\",\"tags\":\"{\\\"key1\\\":\\\"val1\\\",\\\"key2\\\":\\\"val2\\\"}\"}"; assertEquals(expectedSerializedMetric, lines[1]); }