pierrejeambrun commented on code in PR #48239:
URL: https://github.com/apache/airflow/pull/48239#discussion_r2012186659
##########
task-sdk/src/airflow/sdk/execution_time/task_runner.py:
##########
@@ -443,7 +443,12 @@ def get_message(self) -> ReceiveMsgType:
This will block until the message has been received.
"""
- line = self.input.readline()
+ line = None
+
+ # We need to investigate why some lines are empty ("")
+ while not line:
+ line = self.input.readline()
+
Review Comment:
Need investigation. Apparently there are some empty lines logged there that
need to be skipped.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -536,6 +601,21 @@ def get_logger(trigger_id: int) -> WrappedLogger:
@classmethod
def run_in_process(cls):
+ from airflow.sdk.execution_time import task_runner
+
+ # Parse DAG file, send JSON back up!
+ comms_decoder = task_runner.CommsDecoder[ToTriggerRunner,
ToTriggerSupervisor](
+ input=sys.stdin,
+ decoder=TypeAdapter[ToTriggerRunner](ToTriggerRunner),
+ )
+
+ msg = comms_decoder.get_message()
+ if not isinstance(msg, messages.StartTriggerer):
+ raise RuntimeError(f"Required first message to be a
messages.StartTriggerer, it was {msg}")
Review Comment:
setup comms when the triggerer starts
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -662,17 +744,11 @@ async def connect_stdin() -> asyncio.StreamReader:
stdin = await connect_stdin()
- # The first message must be this type, else we can't operate
- line = await stdin.readline()
-
decoder = TypeAdapter[ToTriggerRunner](ToTriggerRunner)
- msg = decoder.validate_json(line)
- if not isinstance(msg, messages.StartTriggerer) or msg.requests_fd <=
0:
- raise RuntimeError(f"First message to triggerer must be
{messages.StartTriggerer.__name__}")
Review Comment:
Remove this check because the startup message is now consume before by the
`run_in_process` method.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -686,8 +762,6 @@ async def connect_stdin() -> asyncio.StreamReader:
self.to_create.append(msg)
elif isinstance(msg, messages.CancelTriggers):
self.to_cancel.extend(msg.ids)
Review Comment:
Here some ConnectionResult message can appear, skipping those. Not sure if
this is the right approach.
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -662,17 +744,11 @@ async def connect_stdin() -> asyncio.StreamReader:
stdin = await connect_stdin()
- # The first message must be this type, else we can't operate
- line = await stdin.readline()
-
decoder = TypeAdapter[ToTriggerRunner](ToTriggerRunner)
- msg = decoder.validate_json(line)
- if not isinstance(msg, messages.StartTriggerer) or msg.requests_fd <=
0:
- raise RuntimeError(f"First message to triggerer must be
{messages.StartTriggerer.__name__}")
writer_transport, writer_protocol = await loop.connect_write_pipe(
lambda: asyncio.streams.FlowControlMixin(loop=loop),
- os.fdopen(msg.requests_fd, "wb"),
+ task_runner.SUPERVISOR_COMMS.request_socket,
Review Comment:
Use the socket already in place
--
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]
