JVMMetricsHandler implements KafkaHandler { @Override public void handle(final ConsumerRecord<String, Bytes> record) { try { JVMMetricCollection metrics = JVMMetricCollection.parseFrom(record.value().get()); if (log.isDebugEnabled()) { log.debug( "Fetched JVM metrics from service[{}] instance[{}] reported.", metrics.getService(), metrics.getServiceInstance() ); } JVMMetricCollection.Builder builder = metrics.toBuilder(); builder.setService(namingLengthControl.formatServiceName(builder.getService())); builder.setServiceInstance(namingLengthControl.formatInstanceName(builder.getServiceInstance())); builder.getMetricsList().forEach(jvmMetric -> { jvmSourceDispatcher.sendMetric(builder.getService(), builder.getServiceInstance(), jvmMetric); }); } catch (InvalidProtocolBufferException e) { log.error("", e); } } JVMMetricsHandler(ModuleManager moduleManager, KafkaFetcherConfig config); @Override void handle(final ConsumerRecord<String, Bytes> record); @Override String getTopic(); @Override String getConsumePartitions();  }
@Test public void testHandler() { long currentTimeMillis = System.currentTimeMillis(); JVMMetric.Builder jvmBuilder = JVMMetric.newBuilder(); jvmBuilder.setTime(currentTimeMillis); jvmBuilder.setCpu(CPU.newBuilder().setUsagePercent(0.98d).build()); jvmBuilder.addAllMemory(Lists.newArrayList(Memory.newBuilder().setInit(10).setUsed(100).setIsHeap(false).build())); jvmBuilder.addAllMemoryPool(Lists.newArrayList(MemoryPool.newBuilder().build())); jvmBuilder.addAllGc(Lists.newArrayList(GC.newBuilder().build())); JVMMetricCollection metrics = JVMMetricCollection.newBuilder() .setService("service") .setServiceInstance("service-instance") .addMetrics(jvmBuilder.build()) .build(); handler.handle(new ConsumerRecord<>(TOPIC_NAME, 0, 0, "", Bytes.wrap(metrics.toByteArray()))); }