Fail jobs when output files exist but stage out fails

pulsar/client/exceptions.py

@@ -6,11 +6,13 @@
 class PulsarClientTransportError(Exception):
     TIMEOUT = 'timeout'
     CONNECTION_REFUSED = 'connection_refused'
+    NOT_200 = 'not_200'
     UNKNOWN = 'unknown'
 
     messages = {
         TIMEOUT: 'Connection timed out',
         CONNECTION_REFUSED: 'Connection refused',
+        NOT_200: 'Response code not 200',
         UNKNOWN: 'Unknown transport error'
     }
 

pulsar/client/staging/down.py

@@ -9,6 +9,7 @@
 )
 
 from ..action_mapper import FileActionMapper
+from ..exceptions import PulsarClientTransportError
 from ..staging import COMMAND_VERSION_FILENAME
 
 log = getLogger(__name__)
@@ -207,6 +208,8 @@ def _collect_output(self, output_type, action, name):
         log.info("collecting output {} with action {}".format(name, action))
         try:
             return self.output_collector.collect_output(self, output_type, action, name)
+        except PulsarClientTransportError:
+            raise
         except Exception as e:
             if _allow_collect_failure(output_type):
                 log.warning(

pulsar/client/transport/curl.py

@@ -67,14 +67,23 @@ def post_file(url, path):
         # pycurl doesn't always produce a great exception for this,
         # wrap it in a better one.
         message = NO_SUCH_FILE_MESSAGE % (path, url)
-        raise Exception(message)
+        raise FileNotFoundError(message)
     c = _new_curl_object_for_url(url)
     c.setopt(c.HTTPPOST, [("file", (c.FORM_FILE, path.encode('ascii')))])
-    c.perform()
+    try:
+        c.perform()
+    except error as exc:
+        raise PulsarClientTransportError(
+            code=_error_curl_to_pulsar(exc.args[0]),
+            message=POST_FAILED_MESSAGE % (url, exc.args[0]),
+            transport_code=exc.args[0],
+            transport_message=exc.args[1])
     status_code = c.getinfo(HTTP_CODE)
     if int(status_code) != 200:
-        message = POST_FAILED_MESSAGE % (url, status_code)
-        raise Exception(message)
+        raise PulsarClientTransportError(
+            code=PulsarClientTransportError.NOT_200,
+            message=POST_FAILED_MESSAGE % (url, status_code),
+            transport_code=status_code)
 
 
 def get_file(url, path):