**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
"""Callback for when the trigger fires.
        This returns immediately. It relies on trigger to throw an exception,
        otherwise it assumes execution was successful.
        """
        job_state = event["job_state"]
        job_id = event["job_id"]
        if job_state == JobStatus.State.ERROR:
            raise AirflowException(f"Job failed:\n{job_id}")
        if job_state == JobStatus.State.CANCELLED:
            raise AirflowException(f"Job was cancelled:\n{job_id}")
        self.log.info("%s completed successfully.", self.task_id)
        return job_id
    def on_kill(self):
        if self.job_id and self.cancel_on_kill:
            self.hook.cancel_job(job_id=self.job_id, project_id=self.project_id, region=self.region)
class DataprocUpdateClusterOperator(GoogleCloudBaseOperator):