**Apache Airflow version**: main

**Environment**: CI

**What happened**:

The `TestLocalTaskJob.test_mark_success_no_kill` test no longer passes on MSSQL. I initially thought it was a race condition, but even after 5 minutes the TI wasn't running.

https://github.com/apache/airflow/blob/36bdfe8d0ef7e5fc428434f8313cf390ee9acc8f/tests/jobs/test_local_task_job.py#L301-L306

I've tracked down that the issue was introduced with #16301 (cc @ephraimbuddy), but I haven't really dug into why.

**How to reproduce it**:

`./breeze --backend mssql tests tests/jobs/test_local_task_job.py`


```
_____________________________________________________________________________________ TestLocalTaskJob.test_mark_success_no_kill _____________________________________________________________________________________
                                                                                                                                                                                                                      
self = <tests.jobs.test_local_task_job.TestLocalTaskJob object at 0x7f54652abf10>                          
                                                                                                                                                                                                                      
    def test_mark_success_no_kill(self):                                                                   
        """                                                                                                                                                                                                           
        Test that ensures that mark_success in the UI doesn't cause                                                                                                                                                   
        the task to fail, and that the task exits                                                                                                                                                                     
        """                                                                                                                                                                                                           
        dagbag = DagBag(                                                                                                                                                                                              
            dag_folder=TEST_DAG_FOLDER,                                                                                                                                                                               
            include_examples=False,                                                                                                                                                                                   
        )                                                                                                  
        dag = dagbag.dags.get('test_mark_success')                                                                                                                                                                    
        task = dag.get_task('task1')                                                                                                                                                                                  
                                                                                                                                                                                                                      
        session = settings.Session()                                                                                                                                                                                  
                                                                                                           
        dag.clear()                                                                                                                                                                                                   
        dag.create_dagrun(                                                                                                                                                                                            
            run_id="test",                                                                                                                                                                                            
            state=State.RUNNING,                                                                                                                                                                                      
            execution_date=DEFAULT_DATE,                                                                                                                                                                              
            start_date=DEFAULT_DATE,                                                                                                                                                                                  
            session=session,                                                                                                                                                                                          
        )                                                                                                  
        ti = TaskInstance(task=task, execution_date=DEFAULT_DATE)                                                                                                                                                     
        ti.refresh_from_db()                                                                                                                                                                                          
        job1 = LocalTaskJob(task_instance=ti, ignore_ti_state=True)                                                                                                                                                   
        process = multiprocessing.Process(target=job1.run)                                                                                                                                                            
        process.start()                                                                                                                                                                                               
        for _ in range(0, 50):                                                                                                                                                                                        
            if ti.state == State.RUNNING:                                                                                                                                                                             
                break                                                                                      
            time.sleep(0.1)                                                                                                                                                                                           
            ti.refresh_from_db()                                                                                                                                                                                      
>       assert State.RUNNING == ti.state                                                                                                                                                                              
E       AssertionError: assert <TaskInstanceState.RUNNING: 'running'> == None                                                                                                                                         
E        +  where <TaskInstanceState.RUNNING: 'running'> = State.RUNNING                                   
E        +  and   None = <TaskInstance: test_mark_success.task1 2016-01-01 00:00:00+00:00 [None]>.state                                                                                                               
                                                                                                                                                                                                                      
tests/jobs/test_local_task_job.py:306: AssertionError                                                                                                                                                                 
------------------------------------------------------------------------------------------------ Captured stderr call ------------------------------------------------------------------------------------------------
INFO  [airflow.models.dagbag.DagBag] Filling up the DagBag from /opt/airflow/tests/dags                                                                                                                               
INFO  [root] class_instance type: <class 'unusual_prefix_5d280a9b385120fec3c40cfe5be04e2f41b6b5e8_test_task_view_type_check.CallableClass'>
INFO  [airflow.models.dagbag.DagBag] File /opt/airflow/tests/dags/test_zip.zip:file_no_airflow_dag.py assumed to contain no DAGs. Skipping.                                                                           
Process Process-1:                                                                                         
Traceback (most recent call last):                                                                                                                                                                                    
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 2336, in _wrap_pool_connect 
    return fn()
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/pool/base.py", line 364, in connect
    return _ConnectionFairy._checkout(self)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/pool/base.py", line 809, in _checkout
    result = pool._dialect.do_ping(fairy.connection)
  File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 575, in do_ping
    cursor.execute(self._dialect_specific_select_one) 
pyodbc.ProgrammingError: ('42000', '[42000] [Microsoft][ODBC Driver 17 for SQL Server][SQL Server]The server failed to resume the transaction. Desc:3400000012. (3971) (SQLExecDirectW)')

(truncated)
```
tests/jobs/test_local_task_job.py
#
#
#
#
import multiprocessing
import os
import signal
import time
import uuid
from datetime import timedelta
from multiprocessing import Lock, Value
from unittest import mock
from unittest.mock import patch
import pytest
from parameterized import parameterized
from airflow import settings
from airflow.exceptions import AirflowException, AirflowFailException
from airflow.executors.sequential_executor import SequentialExecutor
from airflow.jobs.local_task_job import LocalTaskJob
from airflow.jobs.scheduler_job import SchedulerJob
from airflow.models.dag import DAG, DagModel
from airflow.models.dagbag import DagBag
from airflow.models.taskinstance import TaskInstance
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from airflow.task.task_runner.standard_task_runner import StandardTaskRunner