diff --git a/airflow/cli/cli_parser.py b/airflow/cli/cli_parser.py
index a20c0c9c7208f..5c043ccff1d04 100644
--- a/airflow/cli/cli_parser.py
+++ b/airflow/cli/cli_parser.py
@@ -28,6 +28,7 @@
 from tabulate import tabulate_formats
 
 from airflow import settings
+from airflow.cli.commands.legacy_commands import check_legacy_command
 from airflow.configuration import conf
 from airflow.exceptions import AirflowException
 from airflow.executors.executor_loader import ExecutorLoader
@@ -65,6 +66,8 @@ def _check_value(self, action, value):
         if value == 'celery' and executor != ExecutorLoader.CELERY_EXECUTOR:
             message = f'celery subcommand works only with CeleryExecutor, your current executor: {executor}'
             raise ArgumentError(action, message)
+        if action.choices is not None and value not in action.choices:
+            check_legacy_command(action, value)
 
         super()._check_value(action, value)
 
diff --git a/airflow/cli/commands/legacy_commands.py b/airflow/cli/commands/legacy_commands.py
new file mode 100644
index 0000000000000..b2ca64ae08c27
--- /dev/null
+++ b/airflow/cli/commands/legacy_commands.py
@@ -0,0 +1,56 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+from argparse import ArgumentError
+
+COMMAND_MAP = {
+    "worker": "celery worker",
+    "flower": "celery flower",
+    "trigger_dag": "dags trigger",
+    "delete_dag": "dags delete",
+    "show_dag": "dags show",
+    "list_dag": "dags list",
+    "dag_status": "dags status",
+    "backfill": "dags backfill",
+    "list_dag_runs": "dags list_runs",
+    "pause": "dags pause",
+    "unpause": "dags unpause",
+    "test": "tasks test",
+    "clear": "tasks clear",
+    "list_tasks": "tasks list",
+    "task_failed_deps": "tasks failed_deps",
+    "task_state": "tasks state",
+    "run": "tasks run",
+    "render": "tasks render",
+    "initdb": "db init",
+    "resetdb": "db reset",
+    "upgradedb": "db upgrade",
+    "checkdb": "db check",
+    "shell": "db shell",
+    "pool": "pools",
+    "list_users": "users list",
+    "create_user": "users create",
+    "delete_user": "users delete"
+}
+
+
+def check_legacy_command(action, value):
+    """ Checks command value and raise error if value is in removed command """
+    new_command = COMMAND_MAP.get(value)
+    if new_command is not None:
+        msg = f"`airflow {value}` command, has been removed, please use `airflow {new_command}`"
+        raise ArgumentError(action, msg)
diff --git a/tests/cli/commands/test_legacy_commands.py b/tests/cli/commands/test_legacy_commands.py
new file mode 100644
index 0000000000000..42a04ff5bb008
--- /dev/null
+++ b/tests/cli/commands/test_legacy_commands.py
@@ -0,0 +1,61 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+import contextlib
+import io
+import unittest
+from argparse import ArgumentError
+from unittest.mock import MagicMock
+
+from airflow.cli import cli_parser
+from airflow.cli.commands import config_command
+from airflow.cli.commands.legacy_commands import COMMAND_MAP, check_legacy_command
+
+LEGACY_COMMANDS = ["worker", "flower", "trigger_dag", "delete_dag", "show_dag", "list_dag",
+                   "dag_status", "backfill", "list_dag_runs", "pause", "unpause", "test",
+                   "clear", "list_tasks", "task_failed_deps", "task_state", "run",
+                   "render", "initdb", "resetdb", "upgradedb", "checkdb", "shell", "pool",
+                   "list_users", "create_user", "delete_user"]
+
+
+class TestCliDeprecatedCommandsValue(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.parser = cli_parser.get_parser()
+
+    def test_should_display_value(self):
+        with self.assertRaises(SystemExit) as cm_exception, \
+                contextlib.redirect_stderr(io.StringIO()) as temp_stderr:
+            config_command.get_value(self.parser.parse_args(['worker']))
+
+        self.assertEqual(2, cm_exception.exception.code)
+        self.assertIn(
+            "`airflow worker` command, has been removed, "
+            "please use `airflow celery worker`, see help above.",
+            temp_stderr.getvalue().strip()
+        )
+
+    def test_command_map(self):
+        for item in LEGACY_COMMANDS:
+            self.assertIsNotNone(COMMAND_MAP[item])
+
+    def test_check_legacy_command(self):
+        action = MagicMock()
+        with self.assertRaises(ArgumentError) as e:
+            check_legacy_command(action, 'list_users')
+        self.assertEqual(
+            str(e.exception),
+            "argument : `airflow list_users` command, has been removed, please use `airflow users list`")