Tested with both chart 1.1.0rc1 (i.e. main branch r.n.) and 1.0.0 

in Airflow 2.1.0, scheduler does not exit immediately (this was an issue < 2.1.0), but all tasks fail like this:

```
2021-06-08 15:30:17,167] {scheduler_job.py:1241} ERROR - Executor reports task instance <TaskInstance: sqoop_acquisition.terminate_job_flow 2021-06-08 13:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2021-06-08 15:30:17,170] {scheduler_job.py:1241} ERROR - Executor reports task instance <TaskInstance: gsheets.state_mapping.to_s3 2021-06-08 14:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2021-06-08 15:30:17,171] {scheduler_job.py:1241} ERROR - Executor reports task instance <TaskInstance: gsheets.app_event_taxonomy.to_s3 2021-06-08 14:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2021-06-08 15:30:17,172] {scheduler_job.py:1241} ERROR - Executor reports task instance <TaskInstance: gsheets.strain_flavors.to_s3 2021-06-08 14:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2021-06-08 15:30:19,053] {scheduler_job.py:1205} INFO - Executor reports execution of reporting_8hr.dev.cannalytics.feature_duration.sql execution_date=2021-06-08 07:00:00+00:00 exited with status failed for try_number 1
[2021-06-08 15:30:19,125] {scheduler_job.py:1241} ERROR - Executor reports task instance <TaskInstance: reporting_8hr.dev.cannalytics.feature_duration.sql 2021-06-08 07:00:00+00:00 [queued]> finished (failed) although the task says its queued. (Info: None) Was the task killed externally?
[2021-06-08 15:30:23,842] {dagrun.py:429} ERROR - Marking run <DagRun gsheets @ 2021-06-08 14:00:00+00:00: scheduled__2021-06-08T14:00:00+00:00, externally triggered: False> failed
```

@kaxil @jedcunningham  do you see this when you run CKE?  Any suggestions?
airflow/jobs/base_job.py
self.hostname = get_hostname()
        self.executor = executor or ExecutorLoader.get_default_executor()
        self.executor_class = self.executor.__class__.__name__
        self.start_date = timezone.utcnow()
        self.latest_heartbeat = timezone.utcnow()
        if heartrate is not None:
            self.heartrate = heartrate
        self.unixname = getuser()
        self.max_tis_per_query = conf.getint('scheduler', 'max_tis_per_query')
        super().__init__(*args, **kwargs)
    @classmethod
    @provide_session
    def most_recent_job(cls, session=None) -> Optional['BaseJob']:
        """
        Return the most recent job of this type, if any, based on last
        heartbeat received.
        This method should be called on a subclass (i.e. on SchedulerJob) to
        return jobs of that type.
        :param session: Database session
        :rtype: BaseJob or None
        """
        return session.query(cls).order_by(cls.latest_heartbeat.desc()).limit(1).first()
    def is_alive(self, grace_multiplier=2.1):