diff --git a/rally/task/scenario.py b/rally/task/scenario.py
index cb65da082d..c4530f67c6 100644
--- a/rally/task/scenario.py
+++ b/rally/task/scenario.py
@@ -13,7 +13,7 @@
 #    License for the specific language governing permissions and limitations
 #    under the License.
 
-
+import inspect
 import random
 
 import six
@@ -134,10 +134,45 @@ class Scenario(plugin.Plugin,
                 if not result.is_valid:
                     raise exceptions.InvalidScenarioArgument(result.msg)
 
+    @staticmethod
+    def _validate_scenario_args(scenario, name, config):
+        if scenario.is_classbased:
+            # We need initialize scenario class to access instancemethods
+            scenario = scenario().run
+        args, _varargs, varkwargs, defaults = inspect.getargspec(scenario)
+
+        hint_msg = (" Use `rally plugin show --name %s` to display "
+                    "scenario description." % name)
+
+        # scenario always accepts an instance of scenario cls as a first arg
+        missed_args = args[1:]
+        if defaults:
+            # do not require args with default values
+            missed_args = missed_args[:-len(defaults)]
+        if "args" in config:
+            missed_args = set(missed_args) - set(config["args"])
+        if missed_args:
+            msg = ("Argument(s) '%(args)s' should be specified in task config."
+                   "%(hint)s" % {"args": "', '".join(missed_args),
+                                 "hint": hint_msg})
+            raise exceptions.InvalidArgumentsException(msg)
+
+        if varkwargs is None and "args" in config:
+            redundant_args = set(config["args"]) - set(args[1:])
+            if redundant_args:
+                msg = ("Unexpected argument(s) found ['%(args)s'].%(hint)s" %
+                       {"args": "', '".join(redundant_args),
+                        "hint": hint_msg})
+                raise exceptions.InvalidArgumentsException(msg)
+
     @classmethod
     def validate(cls, name, config, admin=None, users=None, deployment=None):
         """Semantic check of benchmark arguments."""
-        validators = Scenario.get(name)._meta_get("validators", default=[])
+        scenario = Scenario.get(name)
+
+        cls._validate_scenario_args(scenario, name, config)
+
+        validators = scenario._meta_get("validators", default=[])
 
         if not validators:
             return
diff --git a/samples/tasks/scenarios/dummy/dummy-random-action.json b/samples/tasks/scenarios/dummy/dummy-random-action.json
index 098749d3a8..8f5e67ca0b 100644
--- a/samples/tasks/scenarios/dummy/dummy-random-action.json
+++ b/samples/tasks/scenarios/dummy/dummy-random-action.json
@@ -2,7 +2,7 @@
     "Dummy.dummy_random_action": [
         {
             "args": {
-                "atomics_num": 5,
+                "actions_num": 5,
                 "sleep_min": 0,
                 "sleep_max": 2
             },
diff --git a/samples/tasks/scenarios/dummy/dummy-random-action.yaml b/samples/tasks/scenarios/dummy/dummy-random-action.yaml
index 45e97a66a5..52a3261aec 100644
--- a/samples/tasks/scenarios/dummy/dummy-random-action.yaml
+++ b/samples/tasks/scenarios/dummy/dummy-random-action.yaml
@@ -2,7 +2,7 @@
   Dummy.dummy_random_action:
     -
       args:
-        atomics_num: 5
+        actions_num: 5
         sleep_min: 0
         sleep_max: 2
       runner:
diff --git a/tests/unit/task/test_scenario.py b/tests/unit/task/test_scenario.py
index 906fda8d1e..aebc9804a2 100644
--- a/tests/unit/task/test_scenario.py
+++ b/tests/unit/task/test_scenario.py
@@ -14,6 +14,7 @@
 #    under the License.
 
 import traceback
+import uuid
 
 import mock
 
@@ -181,6 +182,48 @@ class ScenarioTestCase(test.TestCase):
 
         Testing.validate_user_validators.unregister()
 
+    def test__validate_scenario_args(self):
+
+        class Testing(fakes.FakeScenario):
+            @scenario.configure()
+            def fake_scenario_to_validate_scenario_args(self, arg1, arg2,
+                                                        arg3=None):
+                pass
+
+        name = "Testing.fake_scenario_to_validate_scenario_args"
+        scen = scenario.Scenario.get(name)
+
+        # check case when argument is missed
+        e = self.assertRaises(exceptions.InvalidArgumentsException,
+                              scenario.Scenario._validate_scenario_args,
+                              scen, name, {"args": {"arg1": 3}})
+        self.assertIn("Argument(s) 'arg2' should be specified in task config.",
+                      e.format_message())
+
+        # check case when one argument is redundant
+        e = self.assertRaises(exceptions.InvalidArgumentsException,
+                              scenario.Scenario._validate_scenario_args,
+                              scen, name,
+                              {"args": {"arg1": 1, "arg2": 2, "arg4": 4}})
+
+        self.assertIn("Unexpected argument(s) found ['arg4']",
+                      e.format_message())
+
+    def test__validate_scenario_args_with_class_based_scenario(self):
+        name = "%s.need_dot" % uuid.uuid4()
+
+        @scenario.configure(name=name)
+        class Testing(scenario.Scenario):
+            def run(self, arg):
+                pass
+
+        e = self.assertRaises(exceptions.InvalidArgumentsException,
+                              scenario.Scenario._validate_scenario_args,
+                              Testing, name, {})
+
+        self.assertIn("Argument(s) 'arg' should be specified in task config.",
+                      e.format_message())
+
     def test_sleep_between_invalid_args(self):
         self.assertRaises(exceptions.InvalidArgumentsException,
                           scenario.Scenario().sleep_between, 15, 5)