@Override
    public void handle(final ConsumerRecord<String, Bytes> record) {
        try (Timer ignored = histogram.createTimer()) {
            ThreadSnapshot snapshot = ThreadSnapshot.parseFrom(record.value().get());
            if (log.isDebugEnabled()) {
                log.debug(
                    "Fetched a thread snapshot[{}] from task[{}] reported",
                    snapshot.getTraceSegmentId(),
                    snapshot.getTaskId()
                );
            }

            final ProfileThreadSnapshotRecord snapshotRecord = new ProfileThreadSnapshotRecord();
            snapshotRecord.setTaskId(snapshot.getTaskId());
            snapshotRecord.setSegmentId(snapshot.getTraceSegmentId());
            snapshotRecord.setDumpTime(snapshot.getTime());
            snapshotRecord.setSequence(snapshot.getSequence());
            snapshotRecord.setStackBinary(snapshot.getStack().toByteArray());
            snapshotRecord.setTimeBucket(TimeBucket.getRecordTimeBucket(snapshot.getTime()));

            RecordStreamProcessor.getInstance().in(snapshotRecord);
        } catch (Exception e) {
            errorCounter.inc();
            log.error("handle record failed", e);
        }
    }
A handler deserializes the message of profiling snapshot 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/ProfileTaskHandler.java