Lee-W commented on code in PR #66156:
URL: https://github.com/apache/airflow/pull/66156#discussion_r3180738948
##########
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:
@shahar1 I actually think adding a new exception is fine, as long as it's
meaningful, which seems to be the case here 🤔 Could you please share why we
might not want it? Thanks!
--
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]
