private Map<String, IdTopicConfig> buildCacheTopicConfig(
                CacheType mqType, ProxyClusterObject proxyClusterObject) {
            Map<String, IdTopicConfig> tmpTopicConfigMap = new HashMap<>();
            List<InLongIdObject> inLongIds = proxyClusterObject.getInlongIds();
            if (inLongIds.isEmpty()) {
                return tmpTopicConfigMap;
            }
            int index;
            String[] idItems;
            String groupId;
            String streamId;
            String topicName;
            String tenant;
            String nameSpace;
            for (InLongIdObject idObject : inLongIds) {
                if (idObject == null
                        || StringUtils.isBlank(idObject.getInlongId())
                        || StringUtils.isBlank(idObject.getTopic())) {
                    continue;
                }
                // parse inlong id
                idItems = idObject.getInlongId().split("\\.");
                if (idItems.length == 2) {
                    if (StringUtils.isBlank(idItems[0])) {
                        continue;
                    }
                    groupId = idItems[0].trim();
                    streamId = idItems[1].trim();
                } else {
                    groupId = idObject.getInlongId().trim();
                    streamId = "";
                }
                topicName = idObject.getTopic().trim();
                // change full topic name "pulsar-xxx/test/base_topic_name" to
                // base topic name "base_topic_name"
                index = topicName.lastIndexOf('/');
                if (index >= 0) {
                    topicName = topicName.substring(index + 1).trim();
                }
                tenant = idObject.getParams().getOrDefault(ConfigConstants.KEY_TENANT, "");
                nameSpace = idObject.getParams().getOrDefault(ConfigConstants.KEY_NAMESPACE, "");
                if (StringUtils.isBlank(idObject.getTopic())) {
                    // namespace field must exist and value not be empty,
                    // otherwise it is an illegal configuration item.
                    continue;
                }
                if (mqType.equals(CacheType.TUBE)) {
                    topicName = nameSpace;
                } else if (mqType.equals(CacheType.KAFKA)) {
                    if (topicName.equals(streamId)) {
                        topicName = String.format(Constants.DEFAULT_KAFKA_TOPIC_FORMAT, nameSpace, topicName);
                    }
                }
                IdTopicConfig tmpConfig = new IdTopicConfig();
                tmpConfig.setInlongGroupIdAndStreamId(groupId, streamId);
                tmpConfig.setTenantAndNameSpace(tenant, nameSpace);
                tmpConfig.setTopicName(topicName);
                tmpConfig.setParams(idObject.getParams());
                tmpConfig.setDataType(DataTypeEnum.convert(
                        idObject.getParams().getOrDefault("dataType", DataTypeEnum.TEXT.getType())));
                tmpConfig.setFieldDelimiter(idObject.getParams().getOrDefault("fieldDelimiter", "|"));
                tmpConfig.setFileDelimiter(idObject.getParams().getOrDefault("fileDelimiter", "\n"));
                tmpConfig.setUseExtendedFields(Boolean.valueOf(
                        idObject.getParams().getOrDefault("useExtendedFields", "false")));
                tmpConfig.setMsgWrapType(getPbWrapType(idObject));
                tmpConfig.setV1CompressType(getPbCompressType(idObject));
                tmpTopicConfigMap.put(tmpConfig.getUid(), tmpConfig);
                // add only groupId object for tube
                if (mqType.equals(CacheType.TUBE)
                        && !tmpConfig.getUid().equals(tmpConfig.getInlongGroupId())
                        && tmpTopicConfigMap.get(tmpConfig.getInlongGroupId()) == null) {
                    IdTopicConfig tmpConfig2 = new IdTopicConfig();
                    tmpConfig2.setInlongGroupIdAndStreamId(groupId, "");
                    tmpConfig2.setTenantAndNameSpace(tenant, nameSpace);
                    tmpConfig2.setTopicName(topicName);
                    tmpConfig2.setDataType(tmpConfig.getDataType());
                    tmpConfig2.setFieldDelimiter(tmpConfig.getFieldDelimiter());
                    tmpConfig2.setFileDelimiter(tmpConfig.getFileDelimiter());
                    tmpConfig2.setParams(new HashMap<>(tmpConfig.getParams()));
                    tmpConfig2.setUseExtendedFields(tmpConfig.isUseExtendedFields());
                    tmpConfig2.setMsgWrapType(tmpConfig.getMsgWrapType());
                    tmpConfig2.setV1CompressType(tmpConfig.getV1CompressType());
                    tmpTopicConfigMap.put(tmpConfig2.getUid(), tmpConfig2);
                }
            }
            return tmpTopicConfigMap;
        }
build id2field config based on id2Topic configure

@param mqType  mq cluster type
@param proxyClusterObject  cluster object info

@return   ID to Topic configures
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/ConfigManager.java