diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index 745aeb76848..420f29d4d52 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -27,10 +27,6 @@ from galaxy.exceptions import ObjectInvalid, ObjectNotFound
 from galaxy.jobs.actions.post import ActionBox
 from galaxy.jobs.mapper import JobRunnerMapper
 from galaxy.jobs.runners import BaseJobRunner, JobState
-from galaxy.tools.parameters.output_collect import (
-    LegacyToolProvidedMetadata,
-    ToolProvidedMetadata
-)
 from galaxy.util import safe_makedirs, unicodify
 from galaxy.util.bunch import Bunch
 from galaxy.util.expressions import ExpressionContext
@@ -1184,17 +1180,6 @@ class JobWrapper(object, HasResourceParameters):
 
         # We collect the stderr from tools that write their stderr to galaxy.json
         tool_provided_metadata = self.get_tool_provided_job_metadata()
-        extra_stderr = ''
-        if isinstance(tool_provided_metadata, LegacyToolProvidedMetadata):
-            extra_stderr = [item.get('stderr') for item in tool_provided_metadata.tool_provided_job_metadata if item.get('stderr')]
-        elif isinstance(tool_provided_metadata, ToolProvidedMetadata):
-            extra_stderr = [item.get('stderr') for item in tool_provided_metadata.tool_provided_job_metadata.values() if item.get('stderr')]
-        if extra_stderr:
-            extra_stderr = "\n".join(extra_stderr)
-            if stderr:
-                stderr = "%s\n%s" % (stderr, extra_stderr)
-            else:
-                stderr = extra_stderr
 
         # Check the tool's stdout, stderr, and exit code for errors, but only
         # if the job has not already been marked as having an error.
@@ -1202,7 +1187,7 @@ class JobWrapper(object, HasResourceParameters):
 
         # We set final_job_state to use for dataset management, but *don't* set
         # job.state until after dataset collection to prevent history issues
-        if self.check_tool_output(stdout, stderr, tool_exit_code, job):
+        if self.check_tool_output(stdout, stderr, tool_exit_code, job) and not tool_provided_metadata.has_failed_outputs():
             final_job_state = job.states.OK
         else:
             final_job_state = job.states.ERROR
diff --git a/lib/galaxy/tools/parameters/output_collect.py b/lib/galaxy/tools/parameters/output_collect.py
index cefa90132e9..98c628f67f0 100644
--- a/lib/galaxy/tools/parameters/output_collect.py
+++ b/lib/galaxy/tools/parameters/output_collect.py
@@ -32,6 +32,9 @@ class NullToolProvidedMetadata(object):
     def get_new_dataset_meta_by_basename(self, output_name, basename):
         return {}
 
+    def has_failed_outputs(self):
+        return False
+
 
 class LegacyToolProvidedMetadata(object):
 
@@ -74,6 +77,14 @@ class LegacyToolProvidedMetadata(object):
         log.warning("Called get_new_datasets with legacy tool metadata provider - that is unimplemented.")
         return []
 
+    def has_failed_outputs(self):
+        found_failed = False
+        for meta in self.tool_provided_job_metadata:
+            if meta.get("failed", False):
+                found_failed = True
+
+        return found_failed
+
 
 class ToolProvidedMetadata(object):
 
@@ -112,6 +123,14 @@ class ToolProvidedMetadata(object):
                 extra_kwds.update(element)
                 yield extra_kwds
 
+    def has_failed_outputs(self):
+        found_failed = False
+        for meta in self.tool_provided_job_metadata.values():
+            if meta.get("failed", False):
+                found_failed = True
+
+        return found_failed
+
 
 def collect_dynamic_collections(
     tool,
diff --git a/tools/data_source/upload.py b/tools/data_source/upload.py
index 3011e5f9967..92dd6afd7c1 100644
--- a/tools/data_source/upload.py
+++ b/tools/data_source/upload.py
@@ -43,7 +43,8 @@ def file_err(msg, dataset, json_file):
     json_file.write(dumps(dict(type='dataset',
                                ext='data',
                                dataset_id=dataset.dataset_id,
-                               stderr=msg)) + "\n")
+                               stderr=msg,
+                               failed=True)) + "\n")
     # never remove a server-side upload
     if dataset.type in ('server_dir', 'path_paste'):
         return