Repository: jina-ai/jina
File: jina/serve/runtimes/gateway/graph/topology_graph.py (Python)

Code before:
                        if (
                            endpoint not in executor_endpoint_mapping[self.name]
                            and __default_endpoint__
                            not in executor_endpoint_mapping[self.name]
                        ):
                            return request, metadata

                    if target_executor_pattern is not None and not re.match(
                        target_executor_pattern, self.name
                    ):
                        return request, metadata
                    # otherwise, send to executor and get response
                    try:
                        result = await connection_pool.send_requests_once(
                            requests=self.parts_to_send,
                            deployment=self.name,
                            metadata=self._metadata,
                            head=True,
                            endpoint=endpoint,
                            timeout=self._timeout_send,
                            retries=self._retries,
                        )
                        if isinstance(result, (AioRpcError, InternalNetworkError)):
                            raise result
                        else:
                            resp, metadata = result
                        if WorkerRequestHandler._KEY_RESULT in resp.parameters:
                            # Accumulate results from each Node and then add them to the original
                            self.result_in_params_returned = resp.parameters[
                                WorkerRequestHandler._KEY_RESULT
                            ]
                        request.parameters = request_input_parameters
                        resp.parameters = request_input_parameters
                        self.parts_to_send.clear()
                    except InternalNetworkError as err:
                        self._handle_internalnetworkerror(err)

                    self.end_time = datetime.utcnow()
                    if metadata and 'is-error' in metadata:
                        self.status = resp.header.status
                    return resp, metadata

            return None, {}

        def get_leaf_tasks(
            self,
            connection_pool: GrpcConnectionPool,
            request_to_send: Optional[DataRequest],
            previous_task: Optional[asyncio.Task],
            endpoint: Optional[str] = None,
            executor_endpoint_mapping: Optional[Dict] = None,

Review comment (bug):
correct me if wrong, for me it looks like `send_requests_once` already raises in case of error, so this raising logic is unreachable and not needed.

Code after:
                        if (
                            endpoint not in executor_endpoint_mapping[self.name]
                            and __default_endpoint__
                            not in executor_endpoint_mapping[self.name]
                        ):
                            return request, metadata

                    if target_executor_pattern is not None and not re.match(
                        target_executor_pattern, self.name
                    ):
                        return request, metadata
                    # otherwise, send to executor and get response
                    try:
                        result = await connection_pool.send_requests_once(
                            requests=self.parts_to_send,
                            deployment=self.name,
                            metadata=self._metadata,
                            head=True,
                            endpoint=endpoint,
                            timeout=self._timeout_send,
                            retries=self._retries,
                        )
                        if issubclass(type(result), BaseException):
                            raise result
                        else:
                            resp, metadata = result
                        if WorkerRequestHandler._KEY_RESULT in resp.parameters:
                            # Accumulate results from each Node and then add them to the original
                            self.result_in_params_returned = resp.parameters[
                                WorkerRequestHandler._KEY_RESULT
                            ]
                        request.parameters = request_input_parameters
                        resp.parameters = request_input_parameters
                        self.parts_to_send.clear()
                    except InternalNetworkError as err:
                        self._handle_internalnetworkerror(err)

                    self.end_time = datetime.utcnow()
                    if metadata and 'is-error' in metadata:
                        self.status = resp.header.status
                    return resp, metadata

            return None, {}

        def get_leaf_tasks(
            self,
            connection_pool: GrpcConnectionPool,
            request_to_send: Optional[DataRequest],
            previous_task: Optional[asyncio.Task],
            endpoint: Optional[str] = None,
            executor_endpoint_mapping: Optional[Dict] = None,