Say when not to iterate an operator that finds its remote work by the task instance's identity

Author: dabla

task-sdk/docs/mapped-tasks-vs-iterable-tasks.rst

@@ -463,6 +463,15 @@ Avoid Iterable Tasks when:
 - Sub-tasks need to defer (deferrable operators) or reschedule (reschedule-mode sensors) — a
   sub-task index has no task instance of its own to defer or reschedule against, so either raises
   a non-retryable failure instead of pausing.
+- The operator finds its remote work by the task instance's identity. Every item runs as the one
+  iterated task instance, with its ``dag_id``, ``task_id``, ``run_id`` and ``map_index``: a
+  ``KubernetesPodOperator`` that reattaches (``durable``, or ``reattach_on_restart`` before it)
+  looks a running pod up by those labels and adopts a sibling item's pod, or fails on finding
+  several; an ``EcsRunTaskOperator`` with ``reattach=True`` builds its ``startedBy`` from the same
+  fields. Where the operator takes labels of its own, one that names the item tells the jobs apart:
+  ``labels={"index": "{{ ti.index }}"}`` on the ``KubernetesPodOperator``, as every template of an
+  item renders against the item's own ``ti``. Otherwise turn reattachment off, or map the task with
+  ``.expand()``.
 
 .. tip::
 

task-sdk/tests/task_sdk/definitions/test_iterableoperator.py

@@ -1195,6 +1195,23 @@ def test_a_template_reads_the_indexed_tasks_own_state_store(self):
 
         assert task.task.offset == "7"
 
+    def test_a_template_reads_the_items_own_index(self):
+        """
+        ``{{ ti.index }}`` in a partial kwarg renders per item, so an operator that finds its remote
+        work by labels (a ``KubernetesPodOperator`` that reattaches) can be told the item apart.
+        """
+        with DAG("test_dag") as dag:
+            expand_input = ListOfDictsExpandInput([{"arg1": "a"}, {"arg1": "b"}])
+            mapped_op = MockOperator.partial(task_id="indexed_label", dag=dag, arg2="{{ ti.index }}")._expand(
+                expand_input, strict=True, register_with_dag=False
+            )
+            iterable_op = IterableOperator(operator=mapped_op, expand_input=expand_input, dag=dag)
+
+            with mock_context(task=iterable_op) as context:
+                results = sorted(iterable_op.execute(context=context))
+
+        assert results == [("a", "0", None), ("b", "1", None)]
+
     def test_execute_renders_template_fields_off_the_loop_thread(self):
         """
         Rendering a sub-task's template fields may call the supervisor synchronously: an XComArg in