MetricbeatMetricReader extends AbstractReader<RawMetric, Metric> { @Override public Stream<Metric> read(RawMetric rm) { String str = readUTF8(rm.getRawBody()); return Arrays.stream(str.split("\n")) .flatMap(this::readline) .filter(Objects::nonNull); }  @Override Stream<Metric> read(RawMetric rm);  }
@Test public void testRead() { String str = "{\"@timestamp\":\"2018-03-29T08:27:21.200Z\",\"metricset\":{\"name\":\"network\"," + ("\"module\":\"system\",\"rtt\":3487},\"system\":{\"network\":{\"in\":{\"errors\":0,\"dropped\":0," + "\"bytes\":0,\"packets\":0},\"out\":{\"errors\":0,\"dropped\":0,\"packets\":0,\"bytes\":0}," + "\"name\":\"ip_vti0\"}},\"beat\":{\"name\":\"moby\",\"hostname\":\"moby\",\"version\":\"6.2.3\"}}"); RawMetric rm = new RawMetric(); rm.setRawBody(str.getBytes(StandardCharsets.UTF_8)); List<Metric> list = new MetricbeatMetricReader().read(rm).collect(Collectors.toList()); StringBuilder sb = new StringBuilder(); for (Metric m : list) { sb.append(m.toString()); } System.out.println(sb); Assert .assertTrue(sb .toString() .contains( "Metric{name='system.network.out.bytes', value=0.0, timestamp=1522312041200, info='null', tags={name=moby, hostname=moby, version=6.2.3, module=system}}")); }