def METHOD_NAME(
    self,
    dagrun,
    task,
    complete: bool = False,
    task_instance=None,
    task_uuid = None
) -> TaskMetadata:
    extractor = self._get_extractor(task)
    task_info = f'task_type={get_operator_class(task).__name__} ' \
        f'airflow_dag_id={task.dag_id} ' \
        f'task_id={task.task_id} ' \
        f'airflow_run_id={dagrun.run_id} '
    if extractor:
        # Extracting advanced metadata is only possible when extractor for particular operator
        # is defined. Without it, we can't extract any input or output data.
        if task_uuid:
            extractor.set_context("task_uuid", task_uuid)
        try:
            self.log.debug(
                f'Using extractor {extractor.__class__.__name__} {task_info}')
            if complete:
                task_metadata = extractor.extract_on_complete(task_instance)
            else:
                task_metadata = extractor.extract()
            self.log.debug(
                f"Found task metadata for operation {task.task_id}: {task_metadata}"
            )
            if task_metadata:
                if (not task_metadata.inputs) and (not task_metadata.outputs):
                    inlets = task.get_inlet_defs()
                    outlets = task.get_outlet_defs()
                    self.extract_inlets_and_outlets(task_metadata, inlets, outlets)
                return task_metadata
        except Exception as e:
            self.log.exception(
                f'Failed to extract metadata {e} {task_info}',
            )
    else:
        self.log.debug(
            f'Unable to find an extractor. {task_info}')
        # Only include the unkonwnSourceAttribute facet if there is no extractor
        task_metadata = TaskMetadata(
            name=get_job_name(task),
            run_facets={
                "unknownSourceAttribute": UnknownOperatorAttributeRunFacet(
                    unknownItems=[
                        UnknownOperatorInstance(
                            name=get_operator_class(task).__name__,
                            properties={
                                attr: value for attr, value in task.__dict__.items()
                            },
                        )
                    ]
                )
            },
        )
        inlets = task.get_inlet_defs()
        outlets = task.get_outlet_defs()
        self.extract_inlets_and_outlets(task_metadata, inlets, outlets)
        return task_metadata
    return TaskMetadata(name=get_job_name(task))