[storm] branch master updated: [STORM-3666] Validate component name in rebalance command (#3302)

[et...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ethanli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git


The following commit(s) were added to refs/heads/master by this push:
     new e794990  [STORM-3666] Validate component name in rebalance command (#3302)
e794990 is described below

commit e794990f278fa01284daca94f0c1c577f056d513
Author: Bipin Prasad <bi...@yahoo.com>
AuthorDate: Thu Sep 17 08:54:10 2020 -0500

    [STORM-3666] Validate component name in rebalance command (#3302)
---
 bin/storm.py                                       |  6 ++---
 .../org/apache/storm/daemon/nimbus/Nimbus.java     | 31 +++++++++++++++++-----
 2 files changed, 27 insertions(+), 10 deletions(-)

diff --git a/bin/storm.py b/bin/storm.py
index ac0ebf0..7651a2c 100755
--- a/bin/storm.py
+++ b/bin/storm.py
@@ -781,8 +781,8 @@ def initialize_rebalance_subcommand(subparsers):
     )
 
     sub_parser.add_argument(
-        "-e", "--executors", action="append", default=[],
-        help="change the number of executors for a given component e.g component_name:5"
+        "-e", "--executor", action="append", default=[],
+        help="change the number of executors for a given component e.g. --executor component_name=6"
     )
 
     sub_parser.add_argument(
@@ -1204,7 +1204,7 @@ def deactivate(args):
 
 
 def rebalance(args):
-    for executor in args.executors:
+    for executor in args.executor:
         try:
             _, new_value = executor.split("=")
             new_value = int(new_value)
diff --git a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
index 672ad58..02f7575 100644
--- a/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
+++ b/storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java
@@ -51,6 +51,7 @@ import java.util.Map;
 import java.util.Map.Entry;
 import java.util.NavigableMap;
 import java.util.Set;
+import java.util.TreeSet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -2674,10 +2675,12 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
 
     private Map<String, Object> tryReadTopoConfFromName(final String topoName) throws NotAliveException,
         AuthorizationException, IOException {
-        IStormClusterState state = stormClusterState;
-        String topoId = state.getTopoId(topoName)
-                             .orElseThrow(() -> new WrappedNotAliveException(topoName + " is not alive"));
-        return tryReadTopoConf(topoId, topoCache);
+        return tryReadTopoConf(toTopoId(topoName), topoCache);
+    }
+
+    private StormTopology tryReadTopologyFromName(final String topoName) throws NotAliveException,
+            AuthorizationException, IOException {
+        return tryReadTopology(toTopoId(topoName), topoCache);
     }
 
     @VisibleForTesting
@@ -3374,8 +3377,23 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
             checkAuthorization(topoName, topoConf, operation);
             // Set principal in RebalanceOptions to nil because users are not suppose to set this
             options.set_principal(null);
+            // check if executor counts are correctly specified
+            StormTopology stormTopology = tryReadTopologyFromName(topoName);
+            Set<String> comps = new TreeSet<>();
+            comps.addAll(stormTopology.get_spouts().keySet());
+            comps.addAll(stormTopology.get_bolts().keySet());
             Map<String, Integer> execOverrides = options.is_set_num_executors() ? options.get_num_executors() : Collections.emptyMap();
-            for (Integer value : execOverrides.values()) {
+            for (Map.Entry<String, Integer> e: execOverrides.entrySet()) {
+                String comp = e.getKey();
+                // validate non-system component ids
+                if (!Utils.isSystemId(comp) && !comps.contains(comp)) {
+                    throw new WrappedInvalidTopologyException(
+                            String.format("Invalid component %s for topology %s, valid values are %s",
+                                    comp, topoName, String.join(",", comps))
+                    );
+                }
+                // validate executor count for component
+                Integer value = e.getValue();
                 if (value == null || value <= 0) {
                     throw new WrappedInvalidTopologyException("Number of executors must be greater than 0");
                 }
@@ -4746,8 +4764,7 @@ public class Nimbus implements Iface, Shutdownable, DaemonCommon {
 
                 for (StormBase base : ownerToBasesEntry.getValue()) {
                     try {
-                        String topoId = state.getTopoId(base.get_name())
-                                             .orElseThrow(() -> new WrappedNotAliveException(base.get_name() + " is not alive"));
+                        String topoId = toTopoId(base.get_name());
                         TopologyResources resources = getResourcesForTopology(topoId, base);
                         totalResourcesAggregate = totalResourcesAggregate.add(resources);
                         Assignment ownerAssignment = topoIdToAssignments.get(topoId);