Duplicated outputs of luigi.task.flatten_output

[starhel]

Description
luigi.task.flatten_output does not detect "diamond" problem. If some task is required by more than one wrapper task, its output is duplicated in returned list.

Simple reproduction:

import luigi
from luigi.task import flatten_output
from luigi.util import requires


class TestTask(luigi.ExternalTask):
    def output(self):
        return luigi.LocalTarget("file.txt")
    

@requires(TestTask)
class WrapperOne(luigi.WrapperTask):
    pass
        

@requires(TestTask)
class WrapperTwo(luigi.WrapperTask):
    pass
        

@requires(WrapperOne, WrapperTwo)
class WrapperMaster(luigi.WrapperTask):
    pass
        
        
print([target.path for target in flatten_output(WrapperMaster())])

Actual output (luigi==3.0.3): ['file.txt', 'file.txt']
Expected output: ['file.txt']

[stale]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. If closed, you may revisit when your time allows and reopen! Thank you for your contributions.