potiuk commented on code in PR #53532:
URL: https://github.com/apache/airflow/pull/53532#discussion_r2217780647
##########
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:
Looks good. I approved as it looks cool but @Lee-W -> would be great if you
also take a look closer :)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
