**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.defer(
                trigger=DataprocWorkflowTrigger(
                    name=operation.operation.name,
                    project_id=self.project_id or hook.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",
            )
    def execute_complete(self, context, event=None) -> None:
        """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["status"] in ("failed", "error"):
            self.log.exception("Unexpected error in the operation.")
            raise AirflowException(event["message"])
        self.log.info("Workflow %s completed successfully", event["operation_name"])
class DataprocSubmitJobOperator(GoogleCloudBaseOperator):
    """Submit a job to a cluster.
    :param project_id: Optional. The ID of the Google Cloud project that the job belongs to.
    :param region: Required. The Cloud Dataproc region in which to handle the request.
    :param job: Required. The job resource.
        If a dict is provided, it must be of the same form as the protobuf message
        :class:`~google.cloud.dataproc_v1.types.Job`.
        For the complete list of supported job types and their configurations please take a look here
        https://cloud.google.com/dataproc/docs/reference/rest/v1/projects.regions.jobs
    :param request_id: Optional. A unique id used to identify the request. If the server receives two