diff --git a/changelogs/fragments/73971-non-batch-end_play.yml b/changelogs/fragments/73971-non-batch-end_play.yml
new file mode 100644
index 00000000000..6445d9ff373
--- /dev/null
+++ b/changelogs/fragments/73971-non-batch-end_play.yml
@@ -0,0 +1,2 @@
+bugfixes:
+  - Ensure end_play ends play, not batch (https://github.com/ansible/ansible/issues/73971)
diff --git a/lib/ansible/executor/play_iterator.py b/lib/ansible/executor/play_iterator.py
index 10494bc3410..96f87c2cc95 100644
--- a/lib/ansible/executor/play_iterator.py
+++ b/lib/ansible/executor/play_iterator.py
@@ -216,6 +216,8 @@ class PlayIterator:
             # plays won't try to advance)
             play_context.start_at_task = None
 
+        self.end_play = False
+
     def get_host_state(self, host):
         # Since we're using the PlayIterator to carry forward failed hosts,
         # in the event that a previous host was not in the current inventory
diff --git a/lib/ansible/executor/playbook_executor.py b/lib/ansible/executor/playbook_executor.py
index bfaecba91b3..e0d3a645aa5 100644
--- a/lib/ansible/executor/playbook_executor.py
+++ b/lib/ansible/executor/playbook_executor.py
@@ -23,7 +23,7 @@ import os
 
 from ansible import constants as C
 from ansible import context
-from ansible.executor.task_queue_manager import TaskQueueManager
+from ansible.executor.task_queue_manager import TaskQueueManager, AnsibleEndPlay
 from ansible.module_utils._text import to_text
 from ansible.module_utils.parsing.convert_bool import boolean
 from ansible.plugins.loader import become_loader, connection_loader, shell_loader
@@ -186,7 +186,12 @@ class PlaybookExecutor:
                             # restrict the inventory to the hosts in the serialized batch
                             self._inventory.restrict_to_hosts(batch)
                             # and run it...
-                            result = self._tqm.run(play=play)
+                            try:
+                                result = self._tqm.run(play=play)
+                            except AnsibleEndPlay as e:
+                                result = e.result
+                                break_play = True
+                                break
 
                             # break the play if the result equals the special return code
                             if result & self._tqm.RUN_FAILED_BREAK_PLAY != 0:
diff --git a/lib/ansible/executor/task_queue_manager.py b/lib/ansible/executor/task_queue_manager.py
index eaa262d87ae..48ebf1aaff2 100644
--- a/lib/ansible/executor/task_queue_manager.py
+++ b/lib/ansible/executor/task_queue_manager.py
@@ -81,6 +81,11 @@ class FinalQueue(multiprocessing.queues.Queue):
         )
 
 
+class AnsibleEndPlay(Exception):
+    def __init__(self, result):
+        self.result = result
+
+
 class TaskQueueManager:
 
     '''
@@ -323,6 +328,9 @@ class TaskQueueManager:
         for host_name in iterator.get_failed_hosts():
             self._failed_hosts[host_name] = True
 
+        if iterator.end_play:
+            raise AnsibleEndPlay(play_return)
+
         return play_return
 
     def cleanup(self):
diff --git a/lib/ansible/plugins/strategy/__init__.py b/lib/ansible/plugins/strategy/__init__.py
index 90f7cbc19d7..7f4dafa4c9b 100644
--- a/lib/ansible/plugins/strategy/__init__.py
+++ b/lib/ansible/plugins/strategy/__init__.py
@@ -1161,6 +1161,9 @@ class StrategyBase:
                 for host in self._inventory.get_hosts(iterator._play.hosts):
                     if host.name not in self._tqm._unreachable_hosts:
                         iterator._host_states[host.name].run_state = iterator.ITERATING_COMPLETE
+                        # end_play is used in PlaybookExecutor/TQM to indicate that
+                        # the whole play is supposed to be ended as opposed to just a batch
+                        iterator.end_play = True
                 msg = "ending play"
             else:
                 skipped = True
diff --git a/test/integration/targets/meta_tasks/runme.sh b/test/integration/targets/meta_tasks/runme.sh
index 3ee419cb069..cc951bbba56 100755
--- a/test/integration/targets/meta_tasks/runme.sh
+++ b/test/integration/targets/meta_tasks/runme.sh
@@ -49,4 +49,10 @@ for test_strategy in linear free; do
 
   grep -q "META: ending play" <<< "$out"
   grep -qv 'Failed to end using end_play' <<< "$out"
+
+  out="$(ansible-playbook test_end_play_serial_one.yml -i inventory.yml -e test_strategy=$test_strategy -vv "$@")"
+
+  [ "$(grep -c "Testing end_play on host" <<< "$out" )" -eq 1 ]
+  grep -q "META: ending play" <<< "$out"
+  grep -qv 'Failed to end using end_play' <<< "$out"
 done
diff --git a/test/integration/targets/meta_tasks/test_end_play_serial_one.yml b/test/integration/targets/meta_tasks/test_end_play_serial_one.yml
new file mode 100644
index 00000000000..f838d4a6bcd
--- /dev/null
+++ b/test/integration/targets/meta_tasks/test_end_play_serial_one.yml
@@ -0,0 +1,13 @@
+- name: Testing end_play with serial 1 and strategy {{ test_strategy | default('linear') }}
+  hosts: testhost:testhost2
+  gather_facts: no
+  serial: 1
+  strategy: "{{ test_strategy | default('linear') }}"
+  tasks:
+    - debug:
+        msg: "Testing end_play on host {{ inventory_hostname }}"
+
+    - meta: end_play
+
+    - fail:
+        msg: 'Failed to end using end_play'