Airflow - Get last success instance of the task

I have a need to terminate and start the EMR cluster every 24 hours from Airflow. I have implemented logic to check if the previous task execution date - current execution date =1, then terminate the cluster and create a new one. Otherwise, skip the execution of the EMR creation task. But, I'm seeing a weird scenario where the DAG execution is getting marked as a success but no task is being executed!!! So, to handle this scenario, I'm trying to check for the last success date of the EMR (XCOM will have cluster id) to terminate the cluster before I start a new one.

I'm not successful so far.. any help is appreciated.

DAG Image: The Pink ones indicate skip. If you closely observe, the emr_termination task (last row in the image) after the blank boxes should be green just like for emr_creation task. But, it got skipped and the previous cluster was not terminated

Code:

def emr_termination_trigger(execution_date, prev_execution_date_success, prev_execution_date, **kwargs):
days_diff = (execution_date.date() - prev_execution_date.date()).days
provisioned_product_id, cluster_id = None, None
creation_tsk_status = 'N/A'
try:
    ti = TaskInstance(emr_creation_tsk, execution_date)
    if ti.previous_ti is None:
        print("previous execution of emr creation is not available. feching last 2 execution")
        ti = TaskInstance(emr_creation_tsk, prev_execution_date)
        if ti.previous_ti is None:
            print("last 2 executions are not available. Nothing to terminate")
            creation_tsk_status = None
        else:
            creation_tsk_status = ti.previous_ti.state
            provisioned_product_id, cluster_id = ti.previous_ti.xcom_pull(
                task_ids=emr_creation_tsk.task_id)
    else:
        creation_tsk_status = ti.previous_ti.state
        provisioned_product_id, cluster_id = ti.previous_ti.xcom_pull(task_ids=emr_creation_tsk.task_id)
except:
    pass
print(f"days_diff:{days_diff} - provisioned_prd:{provisioned_product_id} - create_status:{creation_tsk_status}")
if days_diff == 1:
    print("Inside Terminating cluster")
    terminate_cluster = TerminateEMROperator(
        task_id='terminate_cluster',
        provisioned_product_id=provisioned_product_id,
        airflow_conn_id=airflow_conn_id,
        provide_context=True,
        dag=dag)
    try:
        terminate_cluster.execute(context=kwargs)
    except Exception as e:
        print(f"Got exception while terminating the EMR cluster:{str(e)}")