o-nikolas commented on code in PR #38514:
URL: https://github.com/apache/airflow/pull/38514#discussion_r1548204315
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -819,19 +821,20 @@ def _execute(self) -> int | None:
)
try:
- self.job.executor.job_id = self.job.id
- if self.processor_agent:
- self.log.debug("Using PipeCallbackSink as callback sink.")
- self.job.executor.callback_sink = PipeCallbackSink(
- get_sink_pipe=self.processor_agent.get_callbacks_pipe
- )
- else:
- from airflow.callbacks.database_callback_sink import
DatabaseCallbackSink
+ for executor in self.job.executors:
+ executor.job_id = self.job.id
+ if self.processor_agent:
+ self.log.debug("Using PipeCallbackSink as callback sink.")
Review Comment:
> The if-else check should only be done once to begin with.
Fair point, I can do that check just once before the loop
> Since it is not possible for executors to have different callback sinks
Do you mean to say that executors can't have different _types_ of callback
sinks? Or that they should literally share the same in-memory sink object
between all executors?
@uranusjr
--
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: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
