diff --git a/changelogs/fragments/62151-loop_control-until.yml b/changelogs/fragments/62151-loop_control-until.yml
new file mode 100644
index 00000000000..70a17ee47ff
--- /dev/null
+++ b/changelogs/fragments/62151-loop_control-until.yml
@@ -0,0 +1,2 @@
+minor_changes:
+  - loop_control - add a break_when option to to break out of a task loop early based on Jinja2 expressions (https://github.com/ansible/ansible/issues/83442).
diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index a400df6781e..fa09611578f 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -402,6 +402,19 @@ class TaskExecutor:
                     self._final_q.send_callback('v2_runner_item_on_ok', tr)
 
             results.append(res)
+
+            # break loop if break_when conditions are met
+            if self._task.loop_control and self._task.loop_control.break_when:
+                cond = Conditional(loader=self._loader)
+                cond.when = self._task.loop_control.get_validated_value(
+                    'break_when', self._task.loop_control.fattributes.get('break_when'), self._task.loop_control.break_when, templar
+                )
+                if cond.evaluate_conditional(templar, task_vars):
+                    # delete loop vars before exiting loop
+                    del task_vars[loop_var]
+                    break
+
+            # done with loop var, remove for next iteration
             del task_vars[loop_var]
 
             # clear 'connection related' plugin variables for next iteration
diff --git a/lib/ansible/playbook/loop_control.py b/lib/ansible/playbook/loop_control.py
index 8581b1f8b45..f7783f0f3c0 100644
--- a/lib/ansible/playbook/loop_control.py
+++ b/lib/ansible/playbook/loop_control.py
@@ -29,6 +29,7 @@ class LoopControl(FieldAttributeBase):
     pause = NonInheritableFieldAttribute(isa='float', default=0, always_post_validate=True)
     extended = NonInheritableFieldAttribute(isa='bool', always_post_validate=True)
     extended_allitems = NonInheritableFieldAttribute(isa='bool', default=True, always_post_validate=True)
+    break_when = NonInheritableFieldAttribute(isa='list', default=list)
 
     def __init__(self):
         super(LoopControl, self).__init__()
@@ -37,3 +38,10 @@ class LoopControl(FieldAttributeBase):
     def load(data, variable_manager=None, loader=None):
         t = LoopControl()
         return t.load_data(data, variable_manager=variable_manager, loader=loader)
+
+    def _post_validate_break_when(self, attr, value, templar):
+        '''
+        break_when is evaluated after the execution of the loop is complete,
+        and should not be templated during the regular post_validate step.
+        '''
+        return value
diff --git a/test/integration/targets/loop_control/break_when.yml b/test/integration/targets/loop_control/break_when.yml
new file mode 100644
index 00000000000..da3de28937c
--- /dev/null
+++ b/test/integration/targets/loop_control/break_when.yml
@@ -0,0 +1,17 @@
+- hosts: localhost
+  gather_facts: false
+  tasks:
+    - debug: var=item
+      changed_when: false
+      loop:
+        - 1
+        - 2
+        - 3
+        - 4
+      loop_control:
+        break_when: item >= 2
+      register: untiltest
+
+    - assert:
+        that:
+            - untiltest['results']|length == 2
diff --git a/test/integration/targets/loop_control/runme.sh b/test/integration/targets/loop_control/runme.sh
index af065ea0e22..6c71aedd78c 100755
--- a/test/integration/targets/loop_control/runme.sh
+++ b/test/integration/targets/loop_control/runme.sh
@@ -10,3 +10,5 @@ bar_label'
 [ "$(ansible-playbook label.yml "$@" |grep 'item='|sed -e 's/^.*(item=looped_var \(.*\)).*$/\1/')" == "${MATCH}" ]
 
 ansible-playbook extended.yml "$@"
+
+ansible-playbook break_when.yml "$@"
diff --git a/test/units/executor/test_task_executor.py b/test/units/executor/test_task_executor.py
index f562bfa5253..8f95d801dbb 100644
--- a/test/units/executor/test_task_executor.py
+++ b/test/units/executor/test_task_executor.py
@@ -164,9 +164,11 @@ class TestTaskExecutor(unittest.TestCase):
 
         def _copy(exclude_parent=False, exclude_tasks=False):
             new_item = MagicMock()
+            new_item.loop_control = MagicMock(break_when=[])
             return new_item
 
         mock_task = MagicMock()
+        mock_task.loop_control = MagicMock(break_when=[])
         mock_task.copy.side_effect = _copy
 
         mock_play_context = MagicMock()