diff --git a/lib/galaxy/dependencies/pinned-requirements.txt b/lib/galaxy/dependencies/pinned-requirements.txt
index b06e4cc7be7..1bc3767683b 100644
--- a/lib/galaxy/dependencies/pinned-requirements.txt
+++ b/lib/galaxy/dependencies/pinned-requirements.txt
@@ -117,7 +117,7 @@ pkgutil-resolve-name==1.3.10 ; python_version >= "3.7" and python_version < "3.9
 prompt-toolkit==3.0.36 ; python_version >= "3.7" and python_version < "3.12"
 prov==1.5.1 ; python_version >= "3.7" and python_version < "3.12"
 psutil==5.9.4 ; python_version >= "3.7" and python_version < "3.12"
-pulsar-galaxy-lib==0.15.0.dev0 ; python_version >= "3.7" and python_version < "3.12"
+pulsar-galaxy-lib==0.15.0.dev1 ; python_version >= "3.7" and python_version < "3.12"
 pyasn1==0.4.8 ; python_version >= "3.7" and python_version < "3.12"
 pycparser==2.21 ; python_version >= "3.7" and python_version < "3.12"
 pycryptodome==3.16.0 ; python_version >= "3.7" and python_version < "3.12"
diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py
index 39eb547a0fb..2f74366c478 100644
--- a/lib/galaxy/jobs/runners/pulsar.py
+++ b/lib/galaxy/jobs/runners/pulsar.py
@@ -99,6 +99,10 @@ PULSAR_PARAM_SPECS = dict(
         map=specs.to_str_or_none,
         default=None,
     ),
+    amqp_key_prefix=dict(
+        map=specs.to_str_or_none,
+        default=None,
+    ),
     galaxy_url=dict(
         map=specs.to_str_or_none,
         default=None,
@@ -968,7 +972,7 @@ class PulsarMQJobRunner(PulsarJobRunner):
     )
 
 
-DEFAULT_PULSAR_CONTAINER = "galaxy/pulsar-pod-staging:0.15.0.1"
+DEFAULT_PULSAR_CONTAINER = "galaxy/pulsar-pod-staging:0.15.0.2"
 COEXECUTION_DESTENTATION_DEFAULTS = {
     "default_file_action": "remote_transfer",
     "rewrite_parameters": "true",
diff --git a/test/integration/test_coexecution.py b/test/integration/test_coexecution.py
index 27709fcfc9c..4e542699a3b 100644
--- a/test/integration/test_coexecution.py
+++ b/test/integration/test_coexecution.py
@@ -8,7 +8,7 @@ communicate job status updates back.
 For this reason, this test will only work out of the box currently with Docker for Mac,
 rabbitmq installed via Homebrew, and if a fixed port is set for the test.
 
-   GALAXY_TEST_PORT=9234 pytest test/integration/test_kubernetes_staging.py
+   GALAXY_TEST_PORT=9234 pytest test/integration/test_coexecution.py
 
 """
 import os
@@ -186,6 +186,37 @@ tools:
 """
 
 
+TES_CONTAINERIZED_TEMPLATE_CUSTOM_AMQP_KEY = """
+runners:
+  local:
+    load: galaxy.jobs.runners.local:LocalJobRunner
+    workers: 1
+  pulsar_tes:
+    load: galaxy.jobs.runners.pulsar:PulsarTesJobRunner
+    amqp_url: ${amqp_url}
+    amqp_key_prefix: pulsar_foobar34_
+
+execution:
+  default: pulsar_tes_environment
+  environments:
+    pulsar_tes_environment:
+      runner: pulsar_tes
+      tes_url: ${tes_url}
+      docker_enabled: true
+      docker_default_container_id: busybox:ubuntu-14.04
+      pulsar_app_config:
+        message_queue_url: '${container_amqp_url}'
+      env:
+        - name: SOME_ENV_VAR
+          value: '42'
+    local_environment:
+      runner: local
+tools:
+  - id: __DATA_FETCH__
+    environment: local_environment
+"""
+
+
 def tes_job_config(template_str: str, jobs_directory: str) -> str:
     job_conf_template = string.Template(template_str)
     assert AMQP_URL
@@ -312,6 +343,20 @@ class TestTesCoexecutionContainerIntegration(TestCoexecution):
         set_infrastucture_url(config)
 
 
+@integration_util.skip_unless_environ("FUNNEL_SERVER_TARGET")
+class TestTesCoexecutionCustomAmqpKeyContainerIntegration(TestCoexecution):
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config) -> None:
+        config["jobs_directory"] = cls.jobs_directory
+        config["file_path"] = cls.jobs_directory
+        config["job_config_file"] = tes_job_config(TES_CONTAINERIZED_TEMPLATE_CUSTOM_AMQP_KEY, cls.jobs_directory)
+
+        config["default_job_shell"] = "/bin/sh"
+        # Disable tool dependency resolution.
+        config["tool_dependency_dir"] = "none"
+        set_infrastucture_url(config)
+
+
 @integration_util.skip_unless_environ("FUNNEL_SERVER_TARGET")
 class TestTesDependencyResolutionIntegration(TestCoexecution):
     @classmethod
@@ -342,7 +387,7 @@ def to_infrastructure_uri(uri: str) -> str:
     # remap MQ or file server URI hostnames for in-container versions, this is sloppy
     # should actually parse the URI and rebuild with correct host
     # Copied from Pulsar's integraiton tests.
-    infrastructure_host = os.environ.get("GALAXY_TEST_INFRASTRUCTURE_HOST")
+    infrastructure_host = GALAXY_TEST_INFRASTRUCTURE_HOST
     if infrastructure_host == "_PLATFORM_AUTO_":
         system = platform.system()
         if system in ["Darwin", "Windows"]: