**Description**

This operator should cancel running workflow template `on_kill`. This option probably should be configurable because of request_id prameter in a job definition: https://googleapis.dev/python/dataproc/latest/gapic/v1/api.html#google.cloud.dataproc_v1.WorkflowTemplateServiceClient.instantiate_inline_workflow_template
https://googleapis.dev/python/dataproc/latest/gapic/v1/api.html#google.cloud.dataproc_v1.WorkflowTemplateServiceClient.instantiate_workflow_template

Note this will have to cancle the Google Long Running Operation for the workflow template (which will in turn cancel all children jobs and clean up relevant cluster).

**Use case / motivation**

Remove dangling workflow templates

**Related Issues**
#10381
#6371
"job_id": TEST_JOB_ID,
    "region": GCP_REGION,
    "project_id": GCP_PROJECT,
}
DATAPROC_CLUSTER_CONF_EXPECTED = {
    "resource": CLUSTER_NAME,
    "region": GCP_REGION,
    "project_id": GCP_PROJECT,
    "url": DATAPROC_CLUSTER_LINK_DEPRECATED,
}
DATAPROC_CLUSTER_EXPECTED = {
    "cluster_id": CLUSTER_NAME,
    "region": GCP_REGION,
    "project_id": GCP_PROJECT,
}
DATAPROC_WORKFLOW_EXPECTED = {
    "workflow_id": TEST_WORKFLOW_ID,
    "region": GCP_REGION,
    "project_id": GCP_PROJECT,
}
BATCH_ID = "test-batch-id"
BATCH = {
    "spark_batch": {
        "jar_file_uris": ["file:///usr/lib/spark/examples/jars/spark-examples.jar"],
        "main_class": "org.apache.spark.examples.SparkPi",
    },
}
def assert_warning(msg: str, warnings):
    assert any(msg in str(w) for w in warnings)
class DataprocTestBase: