@Override
    public Map<String, String> parse2IdParams(StreamSinkEntity streamSink, List<String> fields,
            DataNodeInfo dataNodeInfo) {
        Map<String, String> param;
        try {
            HashMap<String, Object> streamInfoMap = JsonUtils.parseObject(streamSink.getExtParams(), HashMap.class);
            param = new HashMap<>();
            assert streamInfoMap != null;
            for (String key : streamInfoMap.keySet()) {
                param.put(key, String.valueOf(streamInfoMap.get(key)));
            }
            // put group and stream info
            param.put(KEY_GROUP_ID, streamSink.getInlongGroupId());
            param.put(KEY_STREAM_ID, streamSink.getInlongStreamId());
            return param;
        } catch (Exception e) {
            LOGGER.error(String.format(
                    "cannot parse properties for groupId=%s, streamId=%s, sinkName=%s, the row properties: %s",
                    streamSink.getInlongGroupId(), streamSink.getInlongStreamId(),
                    streamSink.getSinkName(), streamSink.getExtParams()),
                    e);

            return null;
        }
    }
Getting the sink type.

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