**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
)
        self.job_id = new_job_id
        if self.deferrable:
            job = self.hook.get_job(project_id=self.project_id, region=self.region, job_id=self.job_id)
            state = job.status.state
            if state == JobStatus.State.DONE:
                return self.job_id
            elif state == JobStatus.State.ERROR:
                raise AirflowException(f"Job failed:\n{job}")
            elif state == JobStatus.State.CANCELLED:
                raise AirflowException(f"Job was cancelled:\n{job}")
            self.defer(
                trigger=DataprocSubmitTrigger(
                    job_id=self.job_id,
                    project_id=self.project_id,
                    region=self.region,
                    gcp_conn_id=self.gcp_conn_id,
                    impersonation_chain=self.impersonation_chain,
                    polling_interval_seconds=self.polling_interval_seconds,
                ),
                method_name="execute_complete",
            )
        elif not self.asynchronous:
            self.log.info("Waiting for job %s to complete", new_job_id)
            self.hook.wait_for_job(
                job_id=new_job_id, region=self.region, project_id=self.project_id, timeout=self.wait_timeout
            )
            self.log.info("Job %s completed successfully.", new_job_id)
        return self.job_id
    def execute_complete(self, context, event=None) -> None: