Merge pull request #2504 from natefoo/pulsar-client-timeout

[16.04] Bump Pulsar client lib version number and add timeout options to Pulsar runner

lib/galaxy/dependencies/pinned-requirements.txt

@@ -43,7 +43,7 @@ anyjson==0.3.3
 
 # Pulsar requirements
 psutil==4.1.0
-pulsar-galaxy-lib==0.7.0.dev4
+pulsar-galaxy-lib==0.7.0.dev5
 
 # sqlalchemy-migrate and dependencies
 sqlalchemy-migrate==0.10.0

lib/galaxy/jobs/runners/pulsar.py

@@ -18,6 +18,7 @@ from pulsar.client import ClientJobDescription
 from pulsar.client import PulsarOutputs
 from pulsar.client import ClientOutputs
 from pulsar.client import PathMapper
+from pulsar.client import PulsarClientTransportError
 
 import pulsar.core
 
@@ -63,6 +64,10 @@ PULSAR_PARAM_SPECS = dict(
         valid=specs.is_in("urllib", "curl", None),
         default=None
     ),
+    transport_timeout=dict(
+        map=lambda val: None if val == "None" else int(val),
+        default=None,
+    ),
     cache=dict(
         map=specs.to_bool_or_none,
         default=None,
@@ -191,7 +196,7 @@ class PulsarJobRunner( AsynchronousJobRunner ):
             client_manager_kwargs[ "file_cache" ] = None
 
         for kwd in self.runner_params.keys():
-            if kwd.startswith( 'amqp_' ):
+            if kwd.startswith( 'amqp_' ) or kwd.startswith( 'transport_' ):
                 client_manager_kwargs[ kwd ] = self.runner_params[ kwd ]
         self.client_manager = build_client_manager(**client_manager_kwargs)
 
@@ -224,6 +229,9 @@ class PulsarJobRunner( AsynchronousJobRunner ):
         try:
             client = self.get_client_from_state(job_state)
             status = client.get_status()
+        except PulsarClientTransportError as exc:
+            log.error("Communication error with Pulsar server on state check, will retry: %s", exc)
+            return job_state
         except Exception:
             # An orphaned job was put into the queue at app startup, so remote server went down
             # either way we are done I guess.