PreethamSanji commented on issue #66158: URL: https://github.com/apache/airflow/issues/66158#issuecomment-5107956030
@csrinivas2130 you offered to take this back in May — are you still working on it? I don't see a PR, so I'll assume it's free, but happy to hand over or collaborate if you've got something in progress. Before proposing a fix I want to flag that I think this thread now has **two different bugs** in it, and they'll need different changes. **@tirkarthi's original report** — intermittent, correlated with triggerer load, multi-MB strings, `msgspec.json.decode` raising. I think this is in `make_buffered_socket_reader` in `task-sdk/src/airflow/sdk/execution_time/supervisor.py`, and there are two distinct defects: 1. *Unterminated buffer is flushed to the JSON decoder on EOF.* In `cb`, when `recv_into` returns 0, whatever remains in `buffer` is sent to the generator even with no trailing newline. `process_log_messages_from_subprocess` then calls `msgspec.json.decode(line)` on that fragment, raises, logs "Malformed json log line" and `continue`s — message dropped. A trigger emitting a large line and then completing would hit this on close, which matches the reported decode errors. 2. *`process()` rescans from offset 0 on every read.* `buffer.find(b"\n")` restarts at the beginning each time, so a 5MB line arriving in 4096-byte chunks is rescanned ~1250 times over a growing buffer. That slows socket draining exactly when the triggerer is busy, which fits the load correlation and the intermittency. **@mrvarmazyar's report** — I don't think framing explains this one. It's consistent on every run rather than intermittent, it's not tied to large strings, and critically the `.log.trigger.*.log` file is absent or empty rather than present-with-gaps. A corrupted socket read would still leave a file with partial content. Also worth noting `GCSRemoteLogIO` has no streaming handler at all — it's upload-on-close, so an empty/absent object in GCS means the local file was never written. That points upstream of the socket, at trigger log file handler setup under the AF3 supervisor (`trigger_should_wrap` is still set on `GCSTaskHandler`, and I'm not sure that AF2-era wrapping still applies). Could we split these into two issues? I'd like to scope a PR to the two framing defects above, with unit tests driving `make_buffered_socket_reader` over a real `socketpair()` — a multi-MB single-line payload, and a mid-line close — so both are deterministic rather than load-dependent. No DB needed. I'd deliberately keep the transport as-is. The requests channel already uses `length_prefixed_frame_reader` with an explicit 4-byte length prefix and has neither problem; moving the logs channel to the same framing is probably the more durable fix, but it changes a wire protocol between supervisor and subprocess, so I'd rather propose it separately than bundle it here. @mrvarmazyar if you open a separate issue for the GCS case I'm happy to take a look at that too. -- 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]
