### Body

Observed in https://github.com/apache/airflow/actions/runs/5835505313/job/15827357798?pr=33309

```
___________________ ERROR at teardown of test_trigger_firing ___________________

self = <sqlalchemy.future.engine.Connection object at 0x7f92d3327910>

    def _rollback_impl(self):
        assert not self.__branch_from
    
        if self._has_events or self.engine._has_events:
            self.dispatch.rollback(self)
    
        if self._still_open_and_dbapi_connection_is_valid:
            if self._echo:
                if self._is_autocommit_isolation():
                    self._log_info(
                        "ROLLBACK using DBAPI connection.rollback(), "
                        "DBAPI should ignore due to autocommit mode"
                    )
                else:
                    self._log_info("ROLLBACK")
            try:
>               self.engine.dialect.do_rollback(self.connection)

/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py:1062: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7f92d9698650>
dbapi_connection = <sqlalchemy.pool.base._ConnectionFairy object at 0x7f92d33839d0>

    def do_rollback(self, dbapi_connection):
>       dbapi_connection.rollback()
E       sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 140269149157120 and this is thread id 140269532822400.

/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/default.py:683: ProgrammingError

The above exception was the direct cause of the following exception:

    @pytest.fixture(autouse=True, scope="function")
    def close_all_sqlalchemy_sessions():
        from sqlalchemy.orm import close_all_sessions
    
        close_all_sessions()
        yield
>       close_all_sessions()

tests/conftest.py:953: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py:4315: in close_all_sessions
    sess.close()
/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py:1816: in close
    self._close_impl(invalidate=False)
/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py:1858: in _close_impl
    transaction.close(invalidate)
/usr/local/lib/python3.11/site-packages/sqlalchemy/orm/session.py:926: in close
    transaction.close()
/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py:2426: in close
    self._do_close()
/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py:2649: in _do_close
    self._close_impl()
/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py:2635: in _close_impl
    self._connection_rollback_impl()
/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py:2627: in _connection_rollback_impl
    self.connection._rollback_impl()
/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py:1064: in _rollback_impl
    self._handle_dbapi_exception(e, None, None, None, None)
/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py:2134: in _handle_dbapi_exception
    util.raise_(
/usr/local/lib/python3.11/site-packages/sqlalchemy/util/compat.py:211: in raise_
    raise exception
/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/base.py:1062: in _rollback_impl
    self.engine.dialect.do_rollback(self.connection)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <sqlalchemy.dialects.sqlite.pysqlite.SQLiteDialect_pysqlite object at 0x7f92d9698650>
dbapi_connection = <sqlalchemy.pool.base._ConnectionFairy object at 0x7f92d33839d0>

    def do_rollback(self, dbapi_connection):
>       dbapi_connection.rollback()
E       sqlalchemy.exc.ProgrammingError: (sqlite3.ProgrammingError) SQLite objects created in a thread can only be used in that same thread. The object was created in thread id 140269149157120 and this is thread id 140269532822400.
E       (Background on this error at: https://sqlalche.me/e/14/f405)

/usr/local/lib/python3.11/site-packages/sqlalchemy/engine/default.py:683: ProgrammingError
------------------------------ Captured log call -------------------------------
INFO     airflow.jobs.triggerer_job_runner:triggerer_job_runner.py:171 Setting up TriggererHandlerWrapper with handler <FileTaskHandler (NOTSET)>
INFO     airflow.jobs.triggerer_job_runner:triggerer_job_runner.py:227 Setting up logging queue listener with handlers [<LocalQueueHandler (NOTSET)>, <TriggererHandlerWrapper (NOTSET)>]
INFO     airflow.jobs.triggerer_job_runner.TriggerRunner:triggerer_job_runner.py:596 trigger test_dag/test_run/test_ti/-1/1 (ID 1) starting
INFO     airflow.jobs.triggerer_job_runner.TriggerRunner:triggerer_job_runner.py:600 Trigger test_dag/test_run/test_ti/-1/1 (ID 1) fired: TriggerEvent<True>
Level 100 airflow.triggers.testing.SuccessTrigger:triggerer_job_runner.py:633 trigger end
INFO     airflow.jobs.triggerer_job_runner.TriggerRunner:triggerer_job_runner.py:622 trigger test_dag/test_run/test_ti/-1/1 (ID 1) completed
```

### Committer

- [X] I acknowledge that I am a maintainer/committer of the Apache Airflow project.
assert "some_password" not in stdout
def test_is_alive():
    """Checks the heartbeat logic"""
    
    triggerer_job = Job(heartrate=10, state=State.RUNNING)
    assert triggerer_job.is_alive()
    
    triggerer_job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=20)
    assert triggerer_job.is_alive()
    
    triggerer_job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=31)
    assert not triggerer_job.is_alive()
    
    triggerer_job.state = State.SUCCESS
    triggerer_job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=10)
    assert not triggerer_job.is_alive(), "Completed jobs even with recent heartbeat should not be alive"
def test_is_needed(session):
    """Checks the triggerer-is-needed logic"""
    
    triggerer_job = Job(heartrate=10, state=State.RUNNING)
    triggerer_job_runner = TriggererJobRunner(triggerer_job)
    assert triggerer_job_runner.is_needed() is False
    
    trigger = TimeDeltaTrigger(datetime.timedelta(days=7))
    trigger_orm = Trigger.from_object(trigger)
    trigger_orm.id = 1
    session.add(trigger_orm)
    session.commit()
    assert triggerer_job_runner.is_needed() is True
def test_capacity_decode():