STORM-2286 Storm Rebalance command should support arbitrary component parallelism

[danny0405]

For legacy reasons, config TOPOLOGY-TASKS is considered first when schedule a topology, for a component, if user don’t specify TOPOLOGY-TASKS, storm just override it to be equal to component parallelism hint, and schedule based on TOPOLOGY-TASKS later on.

This works for the most cases, but not Rebalance command. Now, when do Rebalance, the StormBase :component->executors attribute will be overridden in Zookeeper which is used to partition component tasks into executors, as we said above, the TOPOLOGY-TASKS is considered here as the real tasks number for components, something goes weird here:

If we override a bigger executor numbers for a component when do rebalance, it just don’t work because smaller TOPOLOGY-TASKS [ not changed since first submitted at all ]is partitioned into bigger number of executors which read from ZooKeeper overridden by Rebalance command, but for smaller task, it works fine.

I see that storm support a command like this now: [storm rebalance topology-name [-w wait-time-secs] [-n new-num-workers] [-e component=parallelism]*] which indicate that user can override a component parallelism freely, i think it’s more sensible to support this and it's meaningless to have a restriction like before.

This is the jira task: STORM-2286

[danny0405]

@erikdw Could you please help me to review this?

[erikdw]

@danny0405 : given that I personally find no value in having this separate task vs. executor distinction that Storm supports (i.e., allowing for a different ratio than 1:1 for tasks to executors), I'm probably not the best person to review this. :) But it would help if you gave some specific examples, instead of only talking abstractly about the problem. Also just FYI, you have a typo in the headline for this PR as well as the storm ticket (Strom --> Storm).

[danny0405]

@erikdw thx for your nice review, i have fixed the headline and ticket title.

For this PR, the ratio is still 1:1 for tasks and executors for a component, but not for the Storm before this PR [for rebalance command].

The key here is that, for the old storm, TOPOLOGY-TASKS is not changed for a component when we do rebalance, we only change the executors number. Thus, when we rebalance a smaller executors number for a component, the old bigger TOPOLOGY-TASKS is partitioned into smaller number of executors, it works [we finally reduce the executors], but some executors will have more than one tasks; When we rebalance a bigger executors number for a component, it doesn't work at all because the old smaller TOPOLOGY-TASKS is partitioned into bigger number of executors, so some executors will not have task assigned and will not start.

For example, for the old storm, if we have a topology named example-topology which contains component like Spout S1[1 executors] -- Bolt B1[2 executors] -- Bolt B2[4 executors], if we do a command storm rebalance example-topology -e B1=1 it will work, and the only executor for B1 will have 2 tasks inside; but for command storm rebalance example-topology -e B1=4, it doesn't work at all and nothing changes finally.

For this PR, all the commands above work, and the ratio for tasks and executors of a component is always 1:1

@revans2 @erikdw please help me to review this again, thx very much ^_^

[erikdw]

I'm unable to reconcile the statements about the ratio always being 1:1 with the content of your "For example" paragraph. Those examples sound like you have a different number of executors than tasks. Am I misreading them?

In any case, I'm not a clojure expert, so you probably wanna get a core contributor (e.g., Bobby Evans, Jungtaek Lim, someone else) to look at this.

However, another Q that comes to mind is: does this behavior also exist in the master branch at HEAD? i.e., the Java-based code. If so then you'll wanna do this change in Java too, right?

[arunmahadevan]

@danny0405 right now rebalance to a higher number of executors is meaningful only if a component was provisioned with more number of tasks than executors during deploy time.

May be there is a reason why the number of tasks as not changed with rebalance. Say if the tasks are maintaining some local state and the inputs to the tasks are keyed on a field and the rebalance alters the number of tasks, the states will not be valid anymore since the same keys will not be mapped to the same tasks anymore.

The idea of dynamically increasing the tasks with rebalance is good, but we need to think through how to handle the key based partitioning use cases.

[danny0405]

@arunmahadevan There are over 2500+ topologies on our cluster, in most cases, we wanna increase component executors too when we add workers to a topology in order to make full use of these workers and also make these executors assigned more evenly among workers. It's awful and time wasting to kill and re-submit the topology just for increasing workers and executors.

As you said, there may be some problem with grouping or something, i need to make sure, thx for your suggestion.

[danny0405]

@HeartSaVioR @arunmahadevan @revans2

I see that now for a rebalancing topology, storm will mark all its slots as free to use for the next round scheduling and is scheduled with other topologies together, this means that it is very possible to schedule to different slots. Here is the snippet:

existing-assignments (into {} (for [tid assigned-topology-ids]
                                        ;; for the topology which wants rebalance (specified by the scratch-topology-id)
                                        ;; we exclude its assignment, meaning that all the slots occupied by its assignment
                                        ;; will be treated as free slot in the scheduler code.
                                        (when (or (nil? scratch-topology-id) (not= tid scratch-topology-id))
                                          {tid (.assignment-info storm-cluster-state tid nil)})))

If we change a component's executors to be smaller, some executors will have more than one tasks which is not accord with the design pattern now. Also the worker containing the changed executors will be restarted.

So why we just kill all its old workers and let them all be reassigned? It just makes no much sense to just keep a few old workers and re-launcher the others.

Personally, i think we should re-launcher all the workers and let the executors be assigned evenly when rebalancing. So i update the PR to change the strategy.

Can you give some suggestions to me? I appreciate it very much.