### Body

Add unit test for changes in https://github.com/apache/airflow/pull/17305

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
tests/jobs/test_backfill_job.py
op4.set_downstream(op5)
            op3.set_downstream(op4)
        runid0 = f'backfill__{DEFAULT_DATE.isoformat()}'
        dag_maker.create_dagrun(run_id=runid0)
        executor = MockExecutor(parallelism=16)
        job = BackfillJob(
            dag=dag,
            executor=executor,
            start_date=DEFAULT_DATE,
            end_date=DEFAULT_DATE + datetime.timedelta(days=2),
        )
        job.run()
        runid1 = f'backfill__{(DEFAULT_DATE + datetime.timedelta(days=1)).isoformat()}'
        runid2 = f'backfill__{(DEFAULT_DATE + datetime.timedelta(days=2)).isoformat()}'
        
        history = executor.history
        assert [sorted(item[-1].key[1:3] for item in batch) for batch in history] == [
            [
                ('leave1', runid0),
                ('leave1', runid1),
                ('leave1', runid2),
                ('leave2', runid0),
                ('leave2', runid1),
                ('leave2', runid2),
            ],
            [('upstream_level_1', runid0), ('upstream_level_1', runid1), ('upstream_level_1', runid2)],
            [('upstream_level_2', runid0), ('upstream_level_2', runid1), ('upstream_level_2', runid2)],
            [('upstream_level_3', runid0), ('upstream_level_3', runid1), ('upstream_level_3', runid2)],
        ]
    def test_backfill_pooled_tasks(self):