Fix GCP Batch runnable ordering: tool (bg) first, sidecar (fg) second

Author: ksuderman

pulsar/client/client.py

@@ -1128,18 +1128,25 @@ def _launch_containers(
         if raw:
             extra_volumes = [v.strip() for v in raw.split(",") if v.strip()]
 
-        runnable = container_command_to_gcp_runnable("pulsar-container", pulsar_submit_container)
-        if extra_volumes:
-            runnable.container.volumes = extra_volumes
-        job.task_groups[0].task_spec.runnables.append(runnable)
-
+        # Order matters: GCP Batch runs runnables sequentially. A background
+        # runnable starts and immediately yields to the next runnable. We need:
+        #   1. Tool (background) — starts polling for command_line file
+        #   2. Sidecar (foreground) — stages inputs, writes command_line, polls
+        #      for return_code, collects outputs, sends AMQP callback
+        # The sidecar must be foreground so it survives after the tool finishes
+        # (GCP Batch kills background runnables when all foreground ones exit).
         if tool_container:
             tool_runnable = container_command_to_gcp_runnable("tool-container", tool_container)
             tool_runnable.background = True
             if extra_volumes:
                 tool_runnable.container.volumes = extra_volumes
             job.task_groups[0].task_spec.runnables.append(tool_runnable)
 
+        runnable = container_command_to_gcp_runnable("pulsar-container", pulsar_submit_container)
+        if extra_volumes:
+            runnable.container.volumes = extra_volumes
+        job.task_groups[0].task_spec.runnables.append(runnable)
+
         job_name = self._job_name
         create_request = gcp_job_request(gcp_job_params, job, job_name)
         client = gcp_client(gcp_job_params.credentials_file)