diff --git a/lib/ansible/executor/task_executor.py b/lib/ansible/executor/task_executor.py
index fb6e1908a7..7943f0d15b 100644
--- a/lib/ansible/executor/task_executor.py
+++ b/lib/ansible/executor/task_executor.py
@@ -458,15 +458,6 @@ class TaskExecutor:
                 vars_copy[self._task.register] = wrap_var(result.copy())
 
             if self._task.async > 0:
-                # the async_wrapper module returns dumped JSON via its stdout
-                # response, so we parse it here and replace the result
-                try:
-                    if 'skipped' in result and result['skipped'] or 'failed' in result and result['failed']:
-                        return result
-                    result = json.loads(result.get('stdout'))
-                except (TypeError, ValueError) as e:
-                    return dict(failed=True, msg=u"The async task did not return valid JSON: %s" % to_unicode(e))
-
                 if self._task.poll > 0:
                     result = self._poll_async_result(result=result, templar=templar)
 
diff --git a/lib/ansible/plugins/action/async.py b/lib/ansible/plugins/action/async.py
index 9b59f64bba..1ea099437d 100644
--- a/lib/ansible/plugins/action/async.py
+++ b/lib/ansible/plugins/action/async.py
@@ -99,4 +99,11 @@ class ActionModule(ActionBase):
 
         result['changed'] = True
 
+        if 'skipped' in result and result['skipped'] or 'failed' in result and result['failed']:
+            return result
+
+        # the async_wrapper module returns dumped JSON via its stdout
+        # response, so we parse it here and replace the result
+        result = self._parse_returned_data(result)
+
         return result
diff --git a/test/integration/Makefile b/test/integration/Makefile
index 73b0af6908..8b1849b502 100644
--- a/test/integration/Makefile
+++ b/test/integration/Makefile
@@ -27,7 +27,7 @@ UNAME := $(shell uname | tr '[:upper:]' '[:lower:]')
 
 all: setup other non_destructive destructive
 
-other: test_test_infra parsing test_var_precedence unicode test_templating_settings environment test_connection includes blocks pull check_mode test_hash test_handlers test_group_by test_vault test_tags test_lookup_paths no_log test_gathering_facts test_binary_modules
+other: test_test_infra parsing test_var_precedence unicode test_templating_settings environment test_connection includes blocks pull check_mode test_hash test_handlers test_group_by test_vault test_tags test_lookup_paths no_log test_gathering_facts test_binary_modules test_async
 
 test_test_infra:
 	# ensure fail/assert work locally and can stop execution with non-zero exit code
@@ -302,3 +302,10 @@ test_binary_modules:
 	cd ..; \
 	rm -rf $$mytmpdir; \
 	ANSIBLE_HOST_KEY_CHECKING=false ansible-playbook test_binary_modules.yml -i $(INVENTORY) -v $(TEST_FLAGS)
+
+test_async:
+	# Verify that extra data before module JSON output during async call is ignored.
+	LC_ALL=bogus ansible-playbook test_async.yml -i $(INVENTORY) -e outputdir=$(TEST_DIR) -v $(TEST_FLAGS)
+	# Verify that the warning exists by examining debug output.
+	ANSIBLE_DEBUG=1 LC_ALL=bogus ansible-playbook test_async.yml -i $(INVENTORY) -e outputdir=$(TEST_DIR) -v $(TEST_FLAGS) \
+	| grep -q 'bash: warning: setlocale: LC_ALL: cannot change locale (bogus)'
diff --git a/test/integration/test_async.yml b/test/integration/test_async.yml
new file mode 100644
index 0000000000..ec370d7c7f
--- /dev/null
+++ b/test/integration/test_async.yml
@@ -0,0 +1,10 @@
+- hosts: testhost3
+  gather_facts: false
+  tasks:
+  # make sure non-JSON data before module output is ignored
+  - name: async ping with invalid locale via ssh
+    ping:
+    async: 3
+    poll: 1
+    register: result
+  - debug: var=result
diff --git a/test/units/executor/test_task_executor.py b/test/units/executor/test_task_executor.py
index b912bc8c59..3f9a614dfb 100644
--- a/test/units/executor/test_task_executor.py
+++ b/test/units/executor/test_task_executor.py
@@ -26,6 +26,7 @@ from ansible.errors import AnsibleError, AnsibleParserError
 from ansible.executor.task_executor import TaskExecutor
 from ansible.playbook.play_context import PlayContext
 from ansible.plugins import action_loader, lookup_loader
+from ansible.parsing.yaml.objects import AnsibleUnicode
 
 from units.mock.loader import DictDataLoader
 
@@ -375,6 +376,7 @@ class TestTaskExecutor(unittest.TestCase):
         # here: on Python 2 comparing MagicMock() > 0 returns True, and the
         # other reason is that if I specify 0 here, the test fails. ;)
         mock_task.async = 1
+        mock_task.poll = 0
 
         mock_play_context = MagicMock()
         mock_play_context.post_validate.return_value = None
@@ -408,11 +410,11 @@ class TestTaskExecutor(unittest.TestCase):
         mock_action.run.return_value = dict(ansible_facts=dict())
         res = te._execute()
 
-        mock_task.changed_when = "1 == 1"
+        mock_task.changed_when = MagicMock(return_value=AnsibleUnicode("1 == 1"))
         res = te._execute()
 
         mock_task.changed_when = None
-        mock_task.failed_when = "1 == 1"
+        mock_task.failed_when = MagicMock(return_value=AnsibleUnicode("1 == 1"))
         res = te._execute()
 
         mock_task.failed_when = None