聊聊storm client的nimbus.seeds參數

序

本文主要研究一下storm client的nimbus.seeds參數

NIMBUS_SEEDS

storm-core-1.1.0-sources.jar!/org/apache/storm/Config.java

/**
     * The host that the master server is running on, added only for backward compatibility,
     * the usage deprecated in favor of nimbus.seeds config.
     */
    @Deprecated
    @isString
    public static final String NIMBUS_HOST = "nimbus.host";

    /**
     * List of seed nimbus hosts to use for leader nimbus discovery.
     */
    @isStringList
    public static final String NIMBUS_SEEDS = "nimbus.seeds";

• 能夠看到這裏廢除了nimbus.host參數，而nimbus.seeds參數主要用於發現nimbus leader

StormSubmitter

storm-core-1.1.0-sources.jar!/org/apache/storm/StormSubmitter.java

public static void submitTopologyAs(String name, Map stormConf, StormTopology topology, SubmitOptions opts, ProgressListener progressListener, String asUser)
            throws AlreadyAliveException, InvalidTopologyException, AuthorizationException, IllegalArgumentException {
        if(!Utils.isValidConf(stormConf)) {
            throw new IllegalArgumentException("Storm conf is not valid. Must be json-serializable");
        }
        stormConf = new HashMap(stormConf);
        stormConf.putAll(Utils.readCommandLineOpts());
        Map conf = Utils.readStormConfig();
        conf.putAll(stormConf);
        stormConf.putAll(prepareZookeeperAuthentication(conf));

        validateConfs(conf, topology);

        Map<String,String> passedCreds = new HashMap<>();
        if (opts != null) {
            Credentials tmpCreds = opts.get_creds();
            if (tmpCreds != null) {
                passedCreds = tmpCreds.get_creds();
            }
        }
        Map<String,String> fullCreds = populateCredentials(conf, passedCreds);
        if (!fullCreds.isEmpty()) {
            if (opts == null) {
                opts = new SubmitOptions(TopologyInitialStatus.ACTIVE);
            }
            opts.set_creds(new Credentials(fullCreds));
        }
        try {
            if (localNimbus!=null) {
                LOG.info("Submitting topology " + name + " in local mode");
                if (opts!=null) {
                    localNimbus.submitTopologyWithOpts(name, stormConf, topology, opts);
                } else {
                    // this is for backwards compatibility
                    localNimbus.submitTopology(name, stormConf, topology);
                }
                LOG.info("Finished submitting topology: " +  name);
            } else {
                String serConf = JSONValue.toJSONString(stormConf);
                try (NimbusClient client = NimbusClient.getConfiguredClientAs(conf, asUser)) {
                    if (topologyNameExists(name, client)) {
                        throw new RuntimeException("Topology with name `" + name + "` already exists on cluster");
                    }

                    // Dependency uploading only makes sense for distributed mode
                    List<String> jarsBlobKeys = Collections.emptyList();
                    List<String> artifactsBlobKeys;

                    DependencyUploader uploader = new DependencyUploader();
                    try {
                        uploader.init();

                        jarsBlobKeys = uploadDependencyJarsToBlobStore(uploader);

                        artifactsBlobKeys = uploadDependencyArtifactsToBlobStore(uploader);
                    } catch (Throwable e) {
                        // remove uploaded jars blobs, not artifacts since they're shared across the cluster
                        uploader.deleteBlobs(jarsBlobKeys);
                        uploader.shutdown();
                        throw e;
                    }

                    try {
                        setDependencyBlobsToTopology(topology, jarsBlobKeys, artifactsBlobKeys);
                        submitTopologyInDistributeMode(name, topology, opts, progressListener, asUser, conf, serConf, client);
                    } catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
                        // remove uploaded jars blobs, not artifacts since they're shared across the cluster
                        // Note that we don't handle TException to delete jars blobs
                        // because it's safer to leave some blobs instead of topology not running
                        uploader.deleteBlobs(jarsBlobKeys);
                        throw e;
                    } finally {
                        uploader.shutdown();
                    }
                }
            }
        } catch(TException e) {
            throw new RuntimeException(e);
        }
        invokeSubmitterHook(name, asUser, conf, topology);

    }

• StormSubmitter的submitTopologyAs經過NimbusClient.getConfiguredClientAs(conf, asUser)建立NimbusClient

NimbusClient

storm-core-1.1.0-sources.jar!/org/apache/storm/utils/NimbusClient.java

public static NimbusClient getConfiguredClientAs(Map conf, String asUser) {
        if (conf.containsKey(Config.STORM_DO_AS_USER)) {
            if (asUser != null && !asUser.isEmpty()) {
                LOG.warn("You have specified a doAsUser as param {} and a doAsParam as config, config will take precedence."
                        , asUser, conf.get(Config.STORM_DO_AS_USER));
            }
            asUser = (String) conf.get(Config.STORM_DO_AS_USER);
        }

        List<String> seeds;
        if(conf.containsKey(Config.NIMBUS_HOST)) {
            LOG.warn("Using deprecated config {} for backward compatibility. Please update your storm.yaml so it only has config {}",
                     Config.NIMBUS_HOST, Config.NIMBUS_SEEDS);
            seeds = Lists.newArrayList(conf.get(Config.NIMBUS_HOST).toString());
        } else {
            seeds = (List<String>) conf.get(Config.NIMBUS_SEEDS);
        }

        for (String host : seeds) {
            int port = Integer.parseInt(conf.get(Config.NIMBUS_THRIFT_PORT).toString());
            NimbusSummary nimbusSummary;
            NimbusClient client = null;
            try {
                client = new NimbusClient(conf, host, port, null, asUser);
                nimbusSummary = client.getClient().getLeader();
                if (nimbusSummary != null) {
                    String leaderNimbus = nimbusSummary.get_host() + ":" + nimbusSummary.get_port();
                    LOG.info("Found leader nimbus : {}", leaderNimbus);
                    if (nimbusSummary.get_host().equals(host) && nimbusSummary.get_port() == port) {
                        NimbusClient ret = client;
                        client = null;
                        return ret;
                    }
                    try {
                        return new NimbusClient(conf, nimbusSummary.get_host(), nimbusSummary.get_port(), null, asUser);
                    } catch (TTransportException e) {
                        throw new RuntimeException("Failed to create a nimbus client for the leader " + leaderNimbus, e);
                    }
                }
            } catch (Exception e) {
                LOG.warn("Ignoring exception while trying to get leader nimbus info from " + host
                                 + ". will retry with a different seed host.", e);
                continue;
            } finally {
                if (client != null) {
                    client.close();
                }
            }
            throw new NimbusLeaderNotFoundException("Could not find a nimbus leader, please try " +
                                                            "again after some time.");
        }
        throw new NimbusLeaderNotFoundException(
                "Could not find leader nimbus from seed hosts " + seeds + ". " +
                        "Did you specify a valid list of nimbus hosts for config " +
                        Config.NIMBUS_SEEDS + "?");
    }

• 這裏仍然兼容NIMBUS_HOST參數，若是有NIMBUS_HOST參數則從中讀取seeds，沒有則從NIMBUS_SEEDS參數獲取
• 以後遍歷seeds，根據每一個seed建立NimbusClient，而後調用client.getClient().getLeader()獲取leader信息，若是獲取成功，則判斷leader是否當前鏈接的seed，若是是則直接返回，若是不是則根據leader的host和port建立新的NimbusClient返回
• 若是nimbusSummary爲null，則會拋出NimbusLeaderNotFoundException("Could not find a nimbus leader, please try again after some time.")
• 若是鏈接leader出現異常，則遍歷下一個seed，進行retry操做，若是全部seed都retry失敗，則跳出循環，最後拋出NimbusLeaderNotFoundException("Could not find leader nimbus from seed hosts " + seeds + ". Did you specify a valid list of nimbus hosts for config nimbus.seeds?")

小結

• 對於storm client來講，nimbus.seeds參數用於client進行尋找nimbus leader，而nimbus.host參數已經被廢棄
• 尋找nimbus leader的過程就是挨個遍歷seeds配置的host，進行鏈接，而後獲取leader的信息，若是獲取成功可是nimbusSummary爲null，則拋出NimbusLeaderNotFoundException("Could not find a nimbus leader, please try again after some time.")。
• 若是有異常則遍歷下一個seed進行retry，若是都不成功，則最後跳出循環，拋出NimbusLeaderNotFoundException("Could not find leader nimbus from seed hosts " + seeds + ". Did you specify a valid list of nimbus hosts for config nimbus.seeds?")

doc

• Setting-up-a-Storm-cluster