### Description

In some cases the Airflow sends `SIGTERM` to the task, here to the `SparkKubernetesOperator`, it needs to send `SIGTERM` also to the corresponding pods/jobs

### Use case/motivation

_No response_

### Related issues

_No response_

### Are you willing to submit a PR?

- [X] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

#
#
#
#
from __future__ import annotations
import json
from unittest.mock import patch
from airflow import DAG
from airflow.models import Connection
from airflow.providers.apache.flink.operators.flink_kubernetes import FlinkKubernetesOperator
from airflow.utils import db, timezone
TEST_VALID_APPLICATION_YAML = """
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flink-sm-ex
  namespace: default
spec:
  image: flink:1.15
  flinkVersion: v1_15
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
    state.savepoints.dir: file:///flink-data/savepoints
    state.checkpoints.dir: file:///flink-data/checkpoints
    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
    high-availability.storageDir: file:///flink-data/ha
  ingress:
    template: "{{name}}.{{namespace}}.flink.k8s.io"
  serviceAccount: flink
  jobManager: