**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.
        """
        if event is None:
            raise AirflowException("Batch failed.")
        state = event["batch_state"]
        batch_id = event["batch_id"]
        self.handle_batch_status(context, state, batch_id)
    def on_kill(self):
        if self.operation:
            self.operation.cancel()
    def handle_batch_status(self, context: Context, state: Batch.State, batch_id: str) -> None:
        
        
        
        link = DATAPROC_BATCH_LINK.format(region=self.region, project_id=self.project_id, batch_id=batch_id)
        if state == Batch.State.FAILED:
            raise AirflowException("Batch job %s failed.  Driver Logs: %s", batch_id, link)
        if state in (Batch.State.CANCELLED, Batch.State.CANCELLING):
            raise AirflowException("Batch job %s was cancelled. Driver logs: %s", batch_id, link)
        if state == Batch.State.STATE_UNSPECIFIED:
            raise AirflowException("Batch job %s unspecified. Driver logs: %s", batch_id, link)
        self.log.info("Batch job %s completed. Driver logs: %s", batch_id, link)
class DataprocDeleteBatchOperator(GoogleCloudBaseOperator):