diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py
index bfad1c028cf..420f29d4d52 100644
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -1178,13 +1178,16 @@ class JobWrapper(object, HasResourceParameters):
             # the tasks failed. So include the stderr, stdout, and exit code:
             return self.fail(job.info, stderr=stderr, stdout=stdout, exit_code=tool_exit_code)
 
+        # We collect the stderr from tools that write their stderr to galaxy.json
+        tool_provided_metadata = self.get_tool_provided_job_metadata()
+
         # 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.
         # The job's stdout and stderr will be set accordingly.
 
         # 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/test/api/test_tools.py b/test/api/test_tools.py
index 8846c4dcbc2..4ef635860b6 100644
--- a/test/api/test_tools.py
+++ b/test/api/test_tools.py
@@ -1366,6 +1366,15 @@ class ToolsTestCase(api.ApiTestCase):
         }
         self._check_combined_mapping_and_subcollection_mapping(history_id, inputs)
 
+    def test_upload_from_invalid_url(self):
+        history_id, dataset_id = self._upload_from_url('https://usegalaxy.org/bla123')
+        dataset_details = self.dataset_populator.get_history_dataset_details(history_id, dataset_id=dataset_id, assert_ok=False)
+        assert dataset_details['state'] == 'error', "expected dataset state to be 'error', but got '%s'" % dataset_details['state']
+
+    def test_upload_from_valid_url(self):
+        history_id, dataset_id = self._upload_from_url('https://usegalaxy.org/api/version')
+        self.dataset_populator.get_history_dataset_details(history_id, dataset_id=dataset_id, assert_ok=True)
+
     def _check_combined_mapping_and_subcollection_mapping(self, history_id, inputs):
         self.dataset_populator.wait_for_history(history_id, assert_ok=True)
         outputs = self._run_and_get_outputs("collection_mixed_param", history_id, inputs)
@@ -1407,6 +1416,19 @@ class ToolsTestCase(api.ApiTestCase):
         else:
             return create_response
 
+    def _upload_from_url(self, url):
+        inputs = {"dbkey": "?",
+                  "file_type": "auto",
+                  "files_0|type":
+                  "upload_dataset",
+                  "files_0|space_to_tab": '',
+                  "files_0|to_posix_lines": "Yes",
+                  "files_0|url_paste": url}
+        history_id = self.dataset_populator.new_history()
+        new_dataset_id = self._run('upload1', history_id=history_id, inputs=inputs).json()['outputs'][0]['id']
+        self.dataset_populator.wait_for_history(history_id, assert_ok=False)
+        return history_id, new_dataset_id
+
     def _upload(self, content, **upload_kwds):
         history_id = self.dataset_populator.new_history()
         new_dataset = self.dataset_populator.new_dataset(history_id, content=content, **upload_kwds)
diff --git a/test/base/populators.py b/test/base/populators.py
index 169b9935cfb..455f2b892d6 100644
--- a/test/base/populators.py
+++ b/test/base/populators.py
@@ -114,15 +114,15 @@ class BaseDatasetPopulator(object):
         payload = self.upload_payload(history_id, content, **kwds)
         run_response = self.tools_post(payload)
         if wait:
-            self.wait_for_tool_run(history_id, run_response)
+            self.wait_for_tool_run(history_id, run_response, kwds.get('assert_ok', True))
         return run_response
 
-    def wait_for_tool_run(self, history_id, run_response):
+    def wait_for_tool_run(self, history_id, run_response, assert_ok=True):
         run = run_response.json()
         assert run_response.status_code == 200, run
         job = run["jobs"][0]
         self.wait_for_job(job["id"])
-        self.wait_for_history(history_id, assert_ok=True)
+        self.wait_for_history(history_id, assert_ok=assert_ok)
         return run_response
 
     def wait_for_history(self, history_id, assert_ok=False, timeout=DEFAULT_TIMEOUT):
diff --git a/test/integration/test_upload_configuration_options.py b/test/integration/test_upload_configuration_options.py
index 1e88759cefd..dc2d91a60ef 100644
--- a/test/integration/test_upload_configuration_options.py
+++ b/test/integration/test_upload_configuration_options.py
@@ -82,9 +82,9 @@ class DefaultBinaryContentFiltersTestCase(BaseCheckUploadContentConfigurationTes
 
     def test_gzipped_html_content_blocked_by_default(self):
         dataset = self.dataset_populator.new_dataset(
-            self.history_id, 'file://%s/bad.html.gz' % TEST_DATA_DIRECTORY, file_type="auto", wait=True
+            self.history_id, 'file://%s/bad.html.gz' % TEST_DATA_DIRECTORY, file_type="auto", wait=True, assert_ok=False
         )
-        dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset)
+        dataset = self.dataset_populator.get_history_dataset_details(self.history_id, dataset=dataset, assert_ok=False)
         assert dataset["file_size"] == 0
 
 
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