diff --git a/airflow/utils/log/json_formatter.py b/airflow/utils/log/json_formatter.py
index 73d461942186c4..f271eba253fd30 100644
--- a/airflow/utils/log/json_formatter.py
+++ b/airflow/utils/log/json_formatter.py
@@ -43,5 +43,16 @@ def usesTime(self):
     def format(self, record):
         super().format(record)
         record_dict = {label: getattr(record, label, None) for label in self.json_fields}
+        if "message" in self.json_fields:
+            msg = record_dict["message"]
+            if record.exc_text:
+                if msg[-1:] != "\n":
+                    msg = msg + "\n"
+                msg = msg + record.exc_text
+            if record.stack_info:
+                if msg[-1:] != "\n":
+                    msg = msg + "\n"
+                msg = msg + self.formatStack(record.stack_info)
+            record_dict["message"] = msg
         merged_record = merge_dicts(record_dict, self.extras)
         return json.dumps(merged_record)
diff --git a/tests/utils/log/test_json_formatter.py b/tests/utils/log/test_json_formatter.py
index b25d11b1a29f49..511e8e0512aeeb 100644
--- a/tests/utils/log/test_json_formatter.py
+++ b/tests/utils/log/test_json_formatter.py
@@ -20,6 +20,7 @@
 Module for all tests airflow.utils.log.json_formatter.JSONFormatter
 """
 import json
+import sys
 import unittest
 from logging import makeLogRecord
 
@@ -63,3 +64,20 @@ def test_format_with_extras(self):
         json_fmt = JSONFormatter(json_fields=["label"], extras={'pod_extra': 'useful_message'})
         # compare as a dicts to not fail on sorting errors
         assert json.loads(json_fmt.format(log_record)) == {"label": "value", "pod_extra": "useful_message"}
+
+    def test_format_with_exception(self):
+        """
+        Test exception is included in the message when using JSONFormatter
+        """
+        try:
+            raise RuntimeError("message")
+        except RuntimeError:
+            exc_info = sys.exc_info()
+
+        log_record = makeLogRecord({"exc_info": exc_info, "message": "Some msg"})
+        json_fmt = JSONFormatter(json_fields=["message"])
+
+        log_fmt = json.loads(json_fmt.format(log_record))
+        assert "message" in log_fmt
+        assert "Traceback (most recent call last)" in log_fmt["message"]
+        assert 'raise RuntimeError("message")' in log_fmt["message"]