evgenii-prusov commented on code in PR #53532:
URL: https://github.com/apache/airflow/pull/53532#discussion_r2217443850
##########
task-sdk/src/airflow/sdk/definitions/mappedoperator.py:
##########
@@ -736,7 +736,19 @@ def unmap(self, resolve: None | Mapping[str, Any]) ->
BaseOperator:
is_setup = kwargs.pop("is_setup", False)
is_teardown = kwargs.pop("is_teardown", False)
on_failure_fail_dagrun = kwargs.pop("on_failure_fail_dagrun",
False)
- kwargs["task_id"] = self.task_id
+
+ # Fix task_id duplication issue: if task_group is present and has
prefix enabled,
+ # strip the prefix from task_id to avoid double prefixing when
BaseOperator.__init__ is called
+ if self.task_group and self.task_group.prefix_group_id:
+ # Strip the task group prefix to avoid double prefixing
+ prefix = f"{self.task_group.group_id}."
+ if self.task_id.startswith(prefix):
+ kwargs["task_id"] = self.task_id[len(prefix) :]
+ else:
+ kwargs["task_id"] = self.task_id
+ else:
+ kwargs["task_id"] = self.task_id
Review Comment:
Thank you for the guidance, @Lee-W. I've made the suggested changes and
executed these tests:
- `breeze run pytest
task-sdk/tests/task_sdk/definitions/test_mappedoperator.py -v`
- `breeze run pytest task-sdk/tests/task_sdk/bases/test_operator.py -v`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
