diff --git a/changelogs/fragments/delegate_to_invalid.yml b/changelogs/fragments/delegate_to_invalid.yml
new file mode 100644
index 00000000000..5eca5f189ba
--- /dev/null
+++ b/changelogs/fragments/delegate_to_invalid.yml
@@ -0,0 +1,2 @@
+bugfixes:
+  - delegate_to when set to an empty or undefined variable will now give a proper error.
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index 60a5b65440b..5b3e5326b6e 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -407,11 +407,7 @@ class TaskExecutor:
         """This method is responsible for effectively pre-validating Task.delegate_to and will
         happen before Task.post_validate is executed
         """
-        delegated_vars, delegated_host_name = self._variable_manager.get_delegated_vars_and_hostname(
-            templar,
-            self._task,
-            variables
-        )
+        delegated_vars, delegated_host_name = self._variable_manager.get_delegated_vars_and_hostname(templar, self._task, variables)
         # At the point this is executed it is safe to mutate self._task,
         # since `self._task` is either a copy referred to by `tmp_task` in `_run_loop`
         # or just a singular non-looped task
diff --git a/lib/ansible/vars/manager.py b/lib/ansible/vars/manager.py
index a332a7f312e..f05ebcdec6c 100644
--- a/lib/ansible/vars/manager.py
+++ b/lib/ansible/vars/manager.py
@@ -537,27 +537,35 @@ class VariableManager:
         delegated_host_name = None
         if task.delegate_to:
             delegated_host_name = templar.template(task.delegate_to, fail_on_undefined=False)
-            delegated_host = self._inventory.get_host(delegated_host_name)
-            if delegated_host is None:
-                for h in self._inventory.get_hosts(ignore_limits=True, ignore_restrictions=True):
-                    # check if the address matches, or if both the delegated_to host
-                    # and the current host are in the list of localhost aliases
-                    if h.address == delegated_host_name:
-                        delegated_host = h
-                        break
-                else:
-                    delegated_host = Host(name=delegated_host_name)
-
-            delegated_vars['ansible_delegated_vars'] = {
-                delegated_host_name: self.get_vars(
-                    play=task.get_play(),
-                    host=delegated_host,
-                    task=task,
-                    include_delegate_to=False,
-                    include_hostvars=True,
-                )
-            }
-            delegated_vars['ansible_delegated_vars'][delegated_host_name]['inventory_hostname'] = variables.get('inventory_hostname')
+
+            # no need to do work if omitted
+            if delegated_host_name != self._omit_token:
+
+                if not delegated_host_name:
+                    raise AnsibleError('Empty hostname produced from delegate_to: "%s"' % task.delegate_to)
+
+                delegated_host = self._inventory.get_host(delegated_host_name)
+                if delegated_host is None:
+                    for h in self._inventory.get_hosts(ignore_limits=True, ignore_restrictions=True):
+                        # check if the address matches, or if both the delegated_to host
+                        # and the current host are in the list of localhost aliases
+                        if h.address == delegated_host_name:
+                            delegated_host = h
+                            break
+                    else:
+                        delegated_host = Host(name=delegated_host_name)
+
+                delegated_vars['ansible_delegated_vars'] = {
+                    delegated_host_name: self.get_vars(
+                        play=task.get_play(),
+                        host=delegated_host,
+                        task=task,
+                        include_delegate_to=False,
+                        include_hostvars=True,
+                    )
+                }
+                delegated_vars['ansible_delegated_vars'][delegated_host_name]['inventory_hostname'] = variables.get('inventory_hostname')
+
         return delegated_vars, delegated_host_name
 
     def _get_delegated_vars(self, play, task, existing_variables):
diff --git a/test/integration/targets/delegate_to/test_delegate_to.yml b/test/integration/targets/delegate_to/test_delegate_to.yml
index 34cfd209423..eb601e02ba6 100644
--- a/test/integration/targets/delegate_to/test_delegate_to.yml
+++ b/test/integration/targets/delegate_to/test_delegate_to.yml
@@ -57,6 +57,25 @@
     - name: remove test file
       file: path={{ output_dir }}/tmp.txt state=absent
 
+    - name: Use omit to thwart delegation
+      ping:
+      delegate_to: "{{ jenkins_install_key_on|default(omit) }}"
+      register: d_omitted
+
+    - name: Use empty to thwart delegation should fail
+      ping:
+      delegate_to: "{{ jenkins_install_key_on }}"
+      when: jenkins_install_key_on != ""
+      vars:
+        jenkins_install_key_on: ''
+      ignore_errors: true
+      register: d_empty
+
+    - name: Ensure previous 2 tests actually did what was expected
+      assert:
+        that:
+          - d_omitted is success
+          - d_empty is failed
 
 - name: verify delegation with per host vars
   hosts: testhost6