[release test] remove JobRunner

[aslonnie]

merge everything into AnyscaleJobRunner

[gemini-code-assist]

Code Review

This pull request refactors the command runners by removing the JobRunner class and merging its functionality into AnyscaleJobRunner. The changes are mostly correct, but there is a critical issue introduced by changing the base class of AnyscaleJobRunner which will cause a NotImplementedError when waiting for nodes. Additionally, the change effectively makes the job run type an alias for anyscale_job, which is a significant behavioral change that should be considered. I've left detailed comments on these points.

[gemini-code-assist]

Changing the base class from JobRunner to CommandRunner introduces a critical issue. The wait_for_nodes method in this class calls super().wait_for_nodes(). Previously, this resolved to JobRunner.wait_for_nodes(). Now, it will resolve to CommandRunner.wait_for_nodes(), which raises a NotImplementedError.

This will break tests that require waiting for nodes. The implementation of wait_for_nodes from JobRunner should be merged into this class. Specifically, the super() call in AnyscaleJobRunner.wait_for_nodes should be replaced with the logic to schedule the wait_cluster.py script, like this:

def wait_for_nodes(self, num_nodes: int, timeout: float = 900):
    self._wait_for_nodes_timeout = timeout
    self.job_manager.cluster_startup_timeout += timeout
    self.run_prepare_command(
        f"python wait_cluster.py {num_nodes} {timeout}", timeout=timeout + 30
    )

[gemini-code-assist]

This change makes run_type="job" an alias for run_type="anyscale_job". This is a significant behavioral change, as "job" previously used JobRunner with FullClusterManager (which manages the cluster lifecycle), and will now use AnyscaleJobRunner with MinimalClusterManager (where the cluster is managed by the Anyscale Job service). If this is intended, consider removing the "job" run type to avoid confusion, as it is now redundant.

[cursor]

Bug: Cluster Node Waiting Logic Missing

The wait_for_nodes method calls super().wait_for_nodes() which raises NotImplementedError in the base CommandRunner class. The old JobRunner implementation added a prepare command to run wait_cluster.py, but this logic was lost during the merge. The method should call self.run_prepare_command(f"python wait_cluster.py {num_nodes} {timeout}", timeout=timeout + 30) instead of calling the parent method.