**Apache Airflow version**: 2.1.0

**What happened**:

We use a python script that reads JSON configs from db to dynamically generate DAGs. Sometimes, when JSON Configs is updated, we expect DAGs to be removed in Airflow.  This doesn't always happen. From my observation, one of the following happens:
1) DAG is removed because the python script no longer generates it.
2) DAG still exists but run history is empty. When trigger the DAG, first task is stuck in `queued` status indefinitely.

**What you expected to happen**:
I expect the right behavior should be 
>1) DAG is removed because the python script no longer generates it.

**How to reproduce it**:
* Create a python script in DAG folder that dynamically generates multiple DAGs
* Execute these dynamically generated DAGs a few times
* Use Airflow Variable to toggle (reduce) the # of DAGs generated
* Examine if number of dynamically generated DAGs in the web UI


airflow/models/dag.py
"""
        Triggers the appropriate callback depending on the value of success, namely the
        on_failure_callback or on_success_callback. This method gets the context of a
        single TaskInstance part of this DagRun and passes that to the callable along
        with a 'reason', primarily to differentiate DagRun failures.
        .. note: The logs end up in
            ``$AIRFLOW_HOME/logs/scheduler/latest/PROJECT/DAG_FILE.py.log``
        :param dagrun: DagRun object
        :param success: Flag to specify if failure or success callback should be called
        :param reason: Completion reason
        :param session: Database session
        """
        callback = self.on_success_callback if success else self.on_failure_callback
        if callback:
            self.log.info('Executing dag callback function: %s', callback)
            tis = dagrun.get_task_instances(session=session)
            ti = tis[-1]  
            ti.task = self.get_task(ti.task_id)
            context = ti.get_template_context(session=session)
            context.update({'reason': reason})
            try:
                callback(context)
            except Exception:
                self.log.exception("failed to invoke dag state update callback")
                Stats.incr("dag.callback_exceptions")
    def get_active_runs(self):