dag_id="test_start_date_scheduling",
                    full_filepath=str(dag_filepath),
                    is_failure_callback=True,
                    run_id=str(i),
                    processor_subdir=str(tmpdir),
                )
                session.add(DbCallbackRequest(callback=callback, priority_weight=i))
        child_pipe, parent_pipe = multiprocessing.Pipe()
        manager = DagFileProcessorManager(
            dag_directory=str(tmpdir),
            max_runs=1,
            processor_timeout=timedelta(days=365),
            signal_conn=child_pipe,
            dag_ids=[],
            pickle_dags=False,
            async_mode=False,
        )
        with create_session() as session:
            self.run_processor_manager_one_loop(manager, parent_pipe)
            assert session.query(DbCallbackRequest).count() == 3
        with create_session() as session:
            self.run_processor_manager_one_loop(manager, parent_pipe)
            assert session.query(DbCallbackRequest).count() == 1
    @conf_vars(
        {
            ("scheduler", "standalone_dag_processor"): "False",
            ("core", "load_examples"): "False",
        }
    )
    def test_fetch_callbacks_from_database_not_standalone(self, tmpdir):