Re: [PR] Fix consistent return response from PubSubPullOperator [airflow]

["VladaZakharova (via GitHub)" <gi...@apache.org>]

VladaZakharova commented on code in PR #66156:
URL: https://github.com/apache/airflow/pull/66156#discussion_r3180526543


##########
providers/google/src/airflow/providers/google/cloud/operators/pubsub.py:
##########
@@ -871,12 +876,22 @@ def execute_complete(self, context: Context, event: dict[str, Any]):
         if event["status"] == "success":
             self.log.info("Sensor pulls messages: %s", event["message"])
             messages_callback = self.messages_callback or self._default_message_callback
-            _return_value = messages_callback(event["message"], context)
+            received_messages = self._convert_to_received_messages(event["message"])
+            _return_value = messages_callback(received_messages, context)
             return _return_value
 
         self.log.info("Sensor failed: %s", event["message"])
         raise AirflowException(event["message"])
 
+    def _convert_to_received_messages(self, messages: Any) -> list[ReceivedMessage]:
+        try:
+            received_messages = [pubsub_v1.types.ReceivedMessage(msg) for msg in messages]
+            return received_messages
+        except Exception as e:
+            raise PubSubMessageTransformException(
+                f"Error converting triggerer event message back to received message format: {e}"
+            )

Review Comment:
   Hi there!
   do we really agree to avoid custom exceptions? AirflowException is forbidden, but here i am not sure we can use something other than custom one



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org