@Override
    public void handle(final ConsumerRecord<String, Bytes> record) {
        try {
            if (record.key().startsWith("register-")) {
                serviceReportProperties(InstanceProperties.parseFrom(record.value().get()));
            } else {
                keepAlive(InstancePingPkg.parseFrom(record.value().get()));
            }
        } catch (Exception e) {
            log.error("handle record failed", e);
        }
    }
A handler deserializes the message of Service Management 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/ServiceManagementHandler.java