@Override
    public ProducerRecord<String, byte[]> parse(KafkaFederationSinkContext context, ProfileEvent event)
            throws IOException {
        String uid = event.getUid();
        KafkaIdConfig idConfig = context.getIdConfig(uid);
        if (idConfig == null) {
            context.addSendResultMetric(event, context.getTaskName(), false, System.currentTimeMillis());
            LOG.error("Can not find the id config:{}", uid);
            return null;
        }
        String delimiter = idConfig.getSeparator();
        byte separator = (byte) delimiter.charAt(0);
        outMsg.reset();
        switch (idConfig.getDataType()) {
            case TEXT:
                currentDate.setTime(event.getRawLogTime());
                String ftime = dateFormat.format(currentDate);
                outMsg.write(ftime.getBytes());
                outMsg.write(separator);
                String extinfo = getExtInfo(event);
                outMsg.write(extinfo.getBytes());
                outMsg.write(separator);
                break;
            case PB:
            case JCE:
            case UNKNOWN:
                break;
            default:
                break;
        }
        outMsg.write(event.getBody());
        byte[] msgContent = outMsg.toByteArray();
        ProducerRecord<String, byte[]> record = new ProducerRecord<>(idConfig.getTopic(), msgContent);
        return record;
    }
parse

@param  context
@param  event
@return             ProducerRecord
@throws IOException
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/sink/kafka/DefaultEvent2KafkaRecordHandler.java