[DO NOT MERGE] Bisect: multitenancy test at 5d2c4e709b (just AFTER 2/2)

[TimothySeah]

Multitenancy release test on master @ 5d2c4e709b (the autoscaler 2/2 PR #63375). Pin/PASS decides whether the regression is in 750ef4e50..5d2c4e709b (everything below, e.g. #63654 BlockEntry, #63680 safe_round) or in 5d2c4e709b..80cc9986e0 (the small remainder).

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.

^{Reviewed by Cursor Bugbot for commit 81b21a3. Configure here.}

[cursor]

Removed memory flag breaks release

High Severity

The --memory CLI option and the memory argument passed into read APIs were removed, but the read_large_parquet release job in release_data_tests.yaml still invokes the script with --memory 3650722201. That run fails at argument parsing and no longer supplies the heap hint the test relied on to avoid OOMs on large Parquet reads.

 

^{Reviewed by Cursor Bugbot for commit 81b21a3. Configure here.}

[cursor]

Py-spy workers lose profiling module

High Severity

worker_scaling_benchmark.py initializes Ray with only benchmark_py_modules(), which now returns just benchmark.py. The weekly worker_scaling release test sets PYSPY_ENABLED=1, which spawns _UDFPySpyProfiler actors on workers that must import profiling.pyspy; that package is no longer shipped via runtime_env after the removed profiling-directory wiring.

Additional Locations (1)

• release/nightly_tests/dataset/benchmark.py#L159-L162

 

^{Reviewed by Cursor Bugbot for commit 81b21a3. Configure here.}

[gemini-code-assist]

Code Review

This pull request introduces a multitenancy variant of the heterogeneous memory batch inference benchmark, including a new cluster configuration and a script to run concurrent pipelines on labeled subclusters while verifying task placement and isolation. It also simplifies several benchmarks by removing unused parameters and simplifying module packaging. The review feedback highlights three key issues: (1) unscheduled tasks lacking a node ID may trigger false-positive placement failures, (2) exceptions in concurrent pipeline threads are silently swallowed and should be propagated to the main thread, and (3) removing the profiling directory from the worker scaling benchmark will break profiling capabilities.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

Pending or unscheduled tasks do not have a node_id assigned yet (i.e., t.node_id is empty or None). If t.node_id is falsy, node_subcluster.get(t.node_id) returns None, which is not in SUBCLUSTERS. This causes the task to fall into the else block, incrementing tasks_on_head and potentially flagging it as a false positive bad_on_head escape. We should skip placement verification for tasks that have not been scheduled on any node yet.

Suggested change

	for t in list_tasks(detail=True, limit=LIST_TASKS_LIMIT):
	total_tasks += 1
	node_sc = node_subcluster.get(t.node_id)
	if node_sc in SUBCLUSTERS:
	for t in list_tasks(detail=True, limit=LIST_TASKS_LIMIT):
	total_tasks += 1
	if not t.node_id:
	continue
	node_sc = node_subcluster.get(t.node_id)
	if node_sc in SUBCLUSTERS:

[gemini-code-assist]

In run_concurrent, the tenant pipelines are executed in separate threads. If an exception occurs within run_pipeline in any of the threads, the exception will be printed to stderr but will not be propagated to the main thread. This can lead to silent failures where the benchmark continues to run, calculates incorrect overhead metrics, or incorrectly passes. We should catch exceptions in the threads and propagate them to the main thread to ensure the benchmark fails fast and explicitly.

def run_concurrent(args: argparse.Namespace) -> Dict[str, float]:
    """Run both tenants on threads concurrently; return per-tenant wall-time."""
    per_tenant: Dict[str, float] = {}
    errors: List[Exception] = []

    def _run(sc: str) -> None:
        try:
            t0 = time.perf_counter()
            run_pipeline(sc, args)
            per_tenant[sc] = time.perf_counter() - t0
        except Exception as e:
            errors.append(e)

    threads = [
        threading.Thread(target=_run, args=(sc,), name=f"pipeline-{sc}")
        for sc in SUBCLUSTERS
    ]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    if errors:
        raise RuntimeError("One or more concurrent pipelines failed.") from errors[0]

    return per_tenant

[gemini-code-assist]

Removing _profiling_dir from py_modules in ray.init will break the worker scaling benchmark when profiling is enabled. The Profiling coordinator spawns _UDFPySpyProfiler actors on worker nodes, which require the profiling package to be available on the workers for deserialization. We should restore shipping the profiling directory to the workers.

Suggested change

	ray.init(runtime_env={"py_modules": benchmark_py_modules()})
	import profiling as _profiling_pkg
	_profiling_dir = os.path.dirname(os.path.abspath(_profiling_pkg.__file__))
	ray.init(runtime_env={"py_modules": benchmark_py_modules() + [_profiling_dir]})