Revert "Update Pulsar for...

Revert "Update Pulsar for https://github.com/galaxyproject/galaxy/commit/96e5aed4df7fdd911ddc0a9d1cecbdafb40ede56."

This reverts commit c4421aa9.

pulsar/client/client.py

@@ -8,7 +8,7 @@ from pulsar.managers.util.pykube_util import (
     galaxy_instance_id,
     Job,
     job_object_dict,
-    produce_k8s_job_prefix,
+    produce_unique_k8s_job_name,
     pull_policy,
     pykube_client_from_dict,
     stop_job,
@@ -479,7 +479,7 @@ class MessageCoexecutionPodJobClient(BaseMessageJobClient):
         base64_message = to_base64_json(launch_params)
         base64_app_conf = to_base64_json(pulsar_app_config)
 
-        k8s_job_prefix = self._k8s_job_prefix
+        job_name = self._k8s_job_name
         params = self.destination_params
 
         pulsar_container_image = self.pulsar_container_image
@@ -527,7 +527,7 @@ class MessageCoexecutionPodJobClient(BaseMessageJobClient):
 
         template = {
             "metadata": {
-                "labels": {"app": k8s_job_prefix},
+                "labels": {"app": job_name},
             },
             "spec": {
                 "volumes": volumes,
@@ -537,13 +537,13 @@ class MessageCoexecutionPodJobClient(BaseMessageJobClient):
         }
         spec = {"template": template}
         spec.update(self._job_spec_params(params))
-        k8s_job_obj = job_object_dict(params, k8s_job_prefix, spec)
+        k8s_job_obj = job_object_dict(params, job_name, spec)
         pykube_client = self._pykube_client
         job = Job(pykube_client, k8s_job_obj)
         job.create()
 
     def kill(self):
-        job_name = self._k8s_job_prefix
+        job_name = self._k8s_job_name
         pykube_client = self._pykube_client
         job = find_job_object_by_name(pykube_client, job_name)
         if job:
@@ -553,7 +553,7 @@ class MessageCoexecutionPodJobClient(BaseMessageJobClient):
             log.info("Attempted to kill k8s job but it is unavailable.")
 
     def job_ip(self):
-        job_name = self._k8s_job_prefix
+        job_name = self._k8s_job_name
         pykube_client = self._pykube_client
         pod = find_pod_object_by_name(pykube_client, job_name)
         if pod:
@@ -572,10 +572,10 @@ class MessageCoexecutionPodJobClient(BaseMessageJobClient):
         return pykube_client_from_dict(self.destination_params)
 
     @property
-    def _k8s_job_prefix(self):
+    def _k8s_job_name(self):
         job_id = self.job_id
-        job_prefix = produce_k8s_job_prefix(app_prefix="pulsar", job_id=job_id, instance_id=self.instance_id)
-        return job_prefix
+        job_name = produce_unique_k8s_job_name(app_prefix="pulsar", job_id=job_id, instance_id=self.instance_id)
+        return job_name
 
     def _job_spec_params(self, params):
         spec = {}