### 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
removed_task_ti.dag_id = subdag.dag_id
        dr.task_instances.append(removed_task_ti)
        session.commit()
        with timeout(seconds=30):
            job.run()
        for task in subdag.tasks:
            instance = (
                session.query(TI)
                .filter(
                    TI.dag_id == subdag.dag_id, TI.task_id == task.task_id, TI.execution_date == DEFAULT_DATE
                )
                .first()
            )
            assert instance is not None
            assert instance.state == State.SUCCESS
        removed_task_ti.refresh_from_db()
        assert removed_task_ti.state == State.REMOVED
        subdag.clear()
        dag.clear()
    def test_update_counters(self, dag_maker):
        with dag_maker(dag_id='test_manage_executor_state', start_date=DEFAULT_DATE) as dag:
            task1 = DummyOperator(task_id='dummy', owner='airflow')
        dr = dag_maker.create_dagrun()
        job = BackfillJob(dag=dag)
        session = settings.Session()
        ti = TI(task1, dr.execution_date)
        ti.refresh_from_db()
        ti_status = BackfillJob._DagRunTaskStatus()
        
        ti.set_state(State.SUCCESS, session)