Solve duplicate task id as a whole

[irainia]

This discussion is started by issue mentioned here. However, rather than solving it partially on the specific bug, it's better to solve it as a whole from end-to-end. Meaning, the user's requirement needs to be reconsidered.

[irainia]

Backround

Task id within a DAG file could be duplicate, and this introduces error during its initialization in Airflow. In Optimus context, within a single DAG, we can have:

• secret
• observability python task
• transformation task
• hook task
• upstream sensor task

Among the above components, both hook task and upstream sensor task can cause duplication task id error if configured in a certain way.

Hook Task

This happens when the user configures more than one hooks of the same type. for example, the user adds two hooks of the same type bq2bq. if this happens, the generated dag will have two hook_bq2bq with the same task id.

Agreed Approach

Currently, we have name only being used to specify the hook. this name is being used as task id. this introduces duplication error if configured in certain ways. to avoid such error, we can introduce name and type. the type can be used to hold the old values, like bq2bq, predator, and so on. the name, however, is the user's defined name. however, they have to be unique within a single job, regardless of the type. of course from the Optimus side there should be business validation on it. and when compiling the dag file, we can use the name as the task id.

Example:

# format: hook_{type}_{name}
hook_bq2bq_notify_slack = SuperKubernetesPodOperator(
  # ...
  task_id='hook_bq2bq_notify_slack'
)
hook_bq2bq_write_status_to_table = SuperKubernetesPodOperator(
  # ...
  task_id='hook_bq2bq_write_status_to_table'
)

Upstream Sensor Task

This happens when the user has more than one internal dependency jobs with the same name, regardless of the project. This is because the way we generate internal upstream sensor task is by using the job name only, without taking into account the project name. So, let's say job1 in project_1 depends on job_2 in project_2 and job_2 in project_3, the sensor task id will be duplicated to job_2.

Agreed Approach

The proper fix was actually implemented. However, due to some business requirements, it was reverted back. The proper way to fix it is to add project name (along with the job name) of the dependencies as task id. Much similar to the one for external dependency. We just need to re-implement it only after all the business requirements allow it. Sensor migration will be needed.