@Override
    protected boolean loadFromFileToHolder() {
        // check meta update setting
        if (!CommonConfigHolder.getInstance().isEnableStartupUsingLocalMetaFile()
                && !ConfigManager.handshakeManagerOk.get()) {
            LOG.warn("Load metadata: StartupUsingLocalMetaFile is false, don't obtain metadata from {}"
                    + " before handshake with Manager", getFileName());
            return false;
        }
        String jsonString = "";
        InLongMetaConfig metaConfig;
        readWriteLock.writeLock().lock();
        try {
            jsonString = loadConfigFromFile();
            if (StringUtils.isBlank(jsonString)) {
                LOG.error("Load metadata: NOT LOADED, changed but empty records, file:{}", getFileName());
                return true;
            }
            try {
                metaConfig = GSON.fromJson(jsonString, InLongMetaConfig.class);
            } catch (Throwable e) {
                LOG.error("Load metadata: NOT LOADED, parse json config failure, file:{}", getFileName(), e);
                return true;
            }
            // check required fields
            ImmutablePair<Boolean, String> paramChkResult = validRequiredFields(metaConfig);
            if (!paramChkResult.getLeft()) {
                LOG.error("Load metadata: NOT LOADED, {}, file:{}",
                        paramChkResult.getRight(), getFileName());
                return true;
            }
            // update cached data
            replaceCacheConfig(metaConfig.getMqType(),
                    metaConfig.getClusterConfigMap(), metaConfig.getIdTopicConfigMap());
            this.dataMd5 = metaConfig.getMd5();
            this.dataStr = jsonString;
            LOG.info("Load metadata: LOADED success, from {}!", getFileName());
            return true;
        } catch (Throwable e) {
            LOG.error("Load metadata: NOT LOADED, load from {} throw exception", getFileName(), e);
            return false;
        } finally {
            if (this.lastSyncVersion.get() == 0) {
                this.lastUpdVersion.set(System.currentTimeMillis());
                this.lastSyncVersion.compareAndSet(0, this.lastUpdVersion.get());
            } else {
                this.lastUpdVersion.set(this.lastSyncVersion.get());
            }
            readWriteLock.writeLock().unlock();
        }
    }
get topic by groupId and streamId
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/config/holder/MetaConfigHolder.java