private List<KuduColumnInfo> getColumnList(SinkInfo sinkInfo) {
        List<StreamSinkFieldEntity> fieldList = sinkFieldMapper.selectBySinkId(sinkInfo.getId());

        // set columns
        List<KuduColumnInfo> columnList = new ArrayList<>();
        for (StreamSinkFieldEntity field : fieldList) {
            if (StringUtils.isNotBlank(field.getExtParams())) {
                KuduColumnInfo kuduColumnInfo = KuduColumnInfo.getFromJson(field.getExtParams());
                CommonBeanUtils.copyProperties(field, kuduColumnInfo, true);
                columnList.add(kuduColumnInfo);
            } else {
                KuduColumnInfo kuduColumnInfo = new KuduColumnInfo();
                CommonBeanUtils.copyProperties(field, kuduColumnInfo, true);
                columnList.add(kuduColumnInfo);
            }
        }

        return columnList;
    }
Create Kudu table according to the sink config
inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/sink/kudu/KuduResourceOperator.java