jason810496 commented on code in PR #74783:
URL: https://github.com/apache/airflow/pull/74783#discussion_r4321177630
##########
task-sdk/src/airflow/sdk/execution_time/supervisor.py:
##########
@@ -2907,7 +2907,15 @@ def process_log_messages_from_subprocess(
if ts := event.get("timestamp"):
# We use msgspec to decode the timestamp as it does it orders of
magnitude quicker than
# datetime.strptime cn
- event["timestamp"] = msgspec.json.decode(f'"{ts}"', type=datetime)
+ try:
+ event["timestamp"] = msgspec.json.decode(f'"{ts}"',
type=datetime)
+ except msgspec.DecodeError:
+ # The subprocess sends the timestamp field, and msgspec cannot
always parse its value. For
+ # example, user code can call self.log.info("...",
timestamp=1697190085).
+ # An uncaught DecodeError would stop the task supervisor
before it reports the
+ # task state, or stop the whole Dag processor during Dag
parsing. Moving the value to
+ # invalid_timestamp lets MaybeTimeStamper add the time the
supervisor received the line.
+ event["invalid_timestamp"] = event.pop("timestamp")
Review Comment:
Instead of setting invalid timestamp, would it be better to just stringfy
the timestamp field?
--
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]
