**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
hook = DataprocHook(gcp_conn_id=self.gcp_conn_id, impersonation_chain=self.impersonation_chain)
        
        if self.batch_id:
            link = DATAPROC_BATCH_LINK.format(
                region=self.region, project_id=self.project_id, batch_id=self.batch_id
            )
            self.log.info("Creating batch %s", self.batch_id)
            self.log.info("Once started, the batch job will be available at %s", link)
        else:
            self.log.info("Starting batch job. The batch ID will be generated since it was not provided.")
        if self.region is None:
            raise AirflowException("Region should be set here")
        try:
            self.operation = hook.create_batch(
                region=self.region,
                project_id=self.project_id,
                batch=self.batch,
                batch_id=self.batch_id,
                request_id=self.request_id,
                retry=self.retry,
                timeout=self.timeout,
                metadata=self.metadata,
            )
            if self.operation is None:
                raise RuntimeError("The operation should be set here!")
            if not self.deferrable:
                if not self.asynchronous:
                    result = hook.wait_for_operation(