@Override
    public void handle(final ConsumerRecord<String, Bytes> record) {
        try (HistogramMetrics.Timer ignored = histogramBatch.createTimer()) {
            MeterDataCollection meterDataCollection = MeterDataCollection.parseFrom(record.value().get());
            MeterProcessor processor = processService.createProcessor();
            meterDataCollection.getMeterDataList().forEach(meterData -> {
                try (HistogramMetrics.Timer ignored2 = histogram.createTimer()) {
                    processor.read(meterData);
                } catch (Exception e) {
                    errorCounter.inc();
                    log.error(e.getMessage(), e);
                }
            });
            processor.process();
        } catch (Exception e) {
            log.error("handle record failed", e);
        }
    }
A handler deserializes the message of meter system data and pushes it to downstream.
oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/provider/handler/MeterServiceHandler.java