Repository: apache/airflow
File: providers/google/src/airflow/providers/google/cloud/hooks/datafusion.py (Python)

Code before:
                "runtimeargs": runtime_args,
                "programType": self.cdap_program_type(pipeline_type=pipeline_type),
                "programId": self.cdap_program_id(pipeline_type=pipeline_type),
            }
        ]
        response = self._cdap_request(url=url, method="POST", body=body)
        self._check_response_status_and_data(
            response, f"Starting a pipeline failed with code {response.status}"
        )
        response_json = json.loads(response.data)
        return response_json[0]["runId"]

    def stop_pipeline(self, pipeline_name: str, instance_url: str, namespace: str = "default") -> None:
        """
        Stop a Cloud Data Fusion pipeline. Works for both batch and stream pipelines.

        :param pipeline_name: Your pipeline name.
        :param instance_url: Endpoint on which the REST APIs is accessible for the instance.
        :param namespace: f your pipeline belongs to a Basic edition instance, the namespace ID
            is always default. If your pipeline belongs to an Enterprise edition instance, you
            can create a namespace.
        """
        url = os.path.join(
            self._base_url(instance_url, namespace),
            quote(pipeline_name),
            "workflows",
            "DataPipelineWorkflow",
            "stop",
        )
        response = self._cdap_request(url=url, method="POST")
        self._check_response_status_and_data(
            response, f"Stopping a pipeline failed with code {response.status}"
        )

    @staticmethod
    def cdap_program_type(pipeline_type: DataFusionPipelineType) -> str:
        """
        Retrieve CDAP Program type depending on the pipeline type.

        :param pipeline_type: Pipeline type.
        """
        program_types = {
            DataFusionPipelineType.BATCH: "workflow",
            DataFusionPipelineType.STREAM: "spark",
        }
        return program_types.get(pipeline_type, "")

    @staticmethod
    def cdap_program_id(pipeline_type: DataFusionPipelineType) -> str:
        """
        Retrieve CDAP Program id depending on the pipeline type.

Review comment (performance):
However @shahar1 I have a suggestion here
The current implemetation of the stop_pipeline hook actually implements https://cdap.atlassian.net/wiki/spaces/DOCS/pages/477560983/Lifecycle+Microservices#Stop-a-Program which actually stops a program not a Run

Maybe we can explore https://cdap.atlassian.net/wiki/spaces/DOCS/pages/477560983/Lifecycle+Microservices#Stop-a-Program-Run here.


Also this was only handling the case of stoping BATCH pipelines, which has been refactored

Code after:
            instance_url,
            "v3",
            "namespaces",
            quote(namespace),
            "start",
        )

        runtime_args = runtime_args or {}
        program_id = self.cdap_program_id(pipeline_type=pipeline_type)

        program_type_value = (
            ProgramType.BATCH.value
            if pipeline_type == DataFusionPipelineType.BATCH
            else ProgramType.STREAM.value
        )

        body = [
            {
                "appId": pipeline_name,
                "programType": program_type_value,
                "programId": program_id,
                "runtimeargs": runtime_args,
            }
        ]

        response: google.auth.transport.Response = self._cdap_request(url=url, method="POST", body=body)

        response_json = {}
        error_message = "Unknown error"

        if response:
            self._check_response_status_and_data(
                response, f"Starting a pipeline failed with code {response.status}"
            )

            response_json = json.loads(response.data)
            if response_json:
                if "runId" in response_json[0]:
                    return response_json[0].get("runId")
                error_message = response_json[0].get("error", error_message)

        raise ValueError(f"Failed to start pipeline '{pipeline_name}'. Error: {error_message}")

    def stop_pipeline(
        self,
        pipeline_name: str,
        instance_url: str,
        namespace: str = "default",
        pipeline_type: DataFusionPipelineType = DataFusionPipelineType.BATCH,
    ) -> None:
        """