dstandish commented on code in PR #40017:
URL: https://github.com/apache/airflow/pull/40017#discussion_r1645055017
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -1844,6 +1895,58 @@ def _executor_to_tis(self, tis: list[TaskInstance]) ->
dict[BaseExecutor, list[T
executor = str(ti.executor)
else:
executor = None
- _executor_to_tis[ExecutorLoader.load_executor(executor)].append(ti)
+
+ if executor_obj := self._try_to_load_executor(executor):
+ _executor_to_tis[executor_obj].append(ti)
+ else:
+ continue
return _executor_to_tis
+
+ def _slots_free_for_tis(self, tis: list[TaskInstance]) ->
set[TaskInstance]:
+ """Return TIs that we have slots available for."""
+ # First get a mapping of executor names to slots they have available
+ executor_to_slots_available: dict[ExecutorName, int] = {}
+ for executor in self.job.executors:
+ # All executors should have a name if they are initted from the
executor_loader. But we need to
+ # check for None to make mypy happy.
Review Comment:
if it's just for mypy, a more self-documenting way to do this is
```python
if TYPE_CHECKING:
assert executor.name
```
then just use the 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]
