@Override
    public void syncField(SinkRequest request, List<StreamField> streamFields) {
        FieldTypeMappingStrategy fieldTypeMappingStrategy = fieldTypeStrategyFactory.getInstance(request.getSinkType());
        if (fieldTypeMappingStrategy == null) {
            LOGGER.info("current sink type ={} not support sync field", request.getSinkType());
            return;
        }
        List<SinkField> sinkFields = request.getSinkFieldList();
        if (sinkFields.size() >= streamFields.size()) {
            return;
        }
        for (int i = sinkFields.size(); i < streamFields.size(); i++) {
            StreamField streamField = streamFields.get(i);
            SinkField sinkField = CommonBeanUtils.copyProperties(streamField, SinkField::new);
            sinkField.setSourceFieldName(streamField.getFieldName());
            sinkField.setSourceFieldType(streamField.getFieldType());
            sinkField.setFieldComment(streamField.getFieldComment());
            sinkField.setFieldName(streamField.getFieldName());
            sinkField.setFieldType(fieldTypeMappingStrategy.getStreamToSinkFieldTypeMapping(streamField.getFieldType())
                    .toLowerCase(Locale.ROOT));
            sinkFields.add(sinkField);
        }
        updateFieldOpt(true, request);
    }
Getting the sink type.

@return sink type string.
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/sink/AbstractSinkOperator.java