VladaZakharova commented on code in PR #66156:
URL: https://github.com/apache/airflow/pull/66156#discussion_r3180532186
##########
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:
there were some discussions in emails, but what was the final decisions
about exceptions?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
