public List<CacheClusterConfig> forkCachedCLusterConfig() {
        List<CacheClusterConfig> result = new ArrayList<>();
        if (mqClusterMap.isEmpty()) {
            return result;
        }
        for (Map.Entry<String, CacheClusterConfig> entry : mqClusterMap.entrySet()) {
            if (entry == null || entry.getKey() == null || entry.getValue() == null) {
                continue;
            }
            CacheClusterConfig config = new CacheClusterConfig();
            config.setClusterName(entry.getValue().getClusterName());
            config.setToken(entry.getValue().getToken());
            config.getParams().putAll(entry.getValue().getParams());
            result.add(config);
        }
        return result;
    }
get topic by groupId and streamId
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/MetaConfigHolder.java