@Override
    @SneakyThrows
    public LogSinkListener parse(final LogData.Builder logData,
                                     final Message extraLog) {
        LogDataBody body = logData.getBody();
        log.setUniqueId(UUID.randomUUID().toString().replace("-", ""));
        // timestamp
        log.setTimestamp(logData.getTimestamp());
        log.setTimeBucket(TimeBucket.getRecordTimeBucket(logData.getTimestamp()));

        // service
        String serviceName = namingControl.formatServiceName(logData.getService());
        String serviceId = IDManager.ServiceID.buildId(serviceName, true);
        log.setServiceId(serviceId);
        // service instance
        if (StringUtil.isNotEmpty(logData.getServiceInstance())) {
            log.setServiceInstanceId(IDManager.ServiceInstanceID.buildId(
                serviceId,
                namingControl.formatInstanceName(logData.getServiceInstance())
            ));
        }
        // endpoint
        if (StringUtil.isNotEmpty(logData.getEndpoint())) {
            String endpointName = namingControl.formatEndpointName(serviceName, logData.getEndpoint());
            log.setEndpointId(IDManager.EndpointID.buildId(serviceId, endpointName));
        }
        // trace
        TraceContext traceContext = logData.getTraceContext();
        if (StringUtil.isNotEmpty(traceContext.getTraceId())) {
            log.setTraceId(traceContext.getTraceId());
        }
        if (StringUtil.isNotEmpty(traceContext.getTraceSegmentId())) {
            log.setTraceSegmentId(traceContext.getTraceSegmentId());
            log.setSpanId(traceContext.getSpanId());
        }
        // content
        if (body.hasText()) {
            log.setContentType(ContentType.TEXT);
            log.setContent(body.getText().getText());
        } else if (body.hasYaml()) {
            log.setContentType(ContentType.YAML);
            log.setContent(body.getYaml().getYaml());
        } else if (body.hasJson()) {
            log.setContentType(ContentType.JSON);
            log.setContent(body.getJson().getJson());
        } else if (extraLog != null) {
            log.setContentType(ContentType.JSON);
            log.setContent(toJSON(extraLog));
        }
        if (logData.getTags().getDataCount() > 0) {
            log.setTagsRawData(logData.getTags().toByteArray());
        }
        log.getTags().addAll(appendSearchableTags(logData));
        return this;
    }
RecordSinkListener forwards the log data to the persistence layer with the query required conditions.
oap-server/analyzer/log-analyzer/src/main/java/org/apache/skywalking/oap/log/analyzer/provider/log/listener/RecordSinkListener.java