Add XCom.deserialize_value to Airflow 1.10.13 (#12328)

closes #11988

airflow/models/xcom.py

@@ -67,17 +67,13 @@ class BaseXCom(Base, LoggingMixin):
     """
     @reconstructor
     def init_on_load(self):
-        enable_pickling = conf.getboolean('core', 'enable_xcom_pickling')
-        if enable_pickling:
+        try:
+            self.value = self.deserialize_value(self)
+        except (UnicodeEncodeError, ValueError):
+            # For backward-compatibility.
+            # Preventing errors in webserver
+            # due to XComs mixed with pickled and unpickled.
             self.value = pickle.loads(self.value)
-        else:
-            try:
-                self.value = json.loads(self.value.decode('UTF-8'))
-            except (UnicodeEncodeError, ValueError):
-                # For backward-compatibility.
-                # Preventing errors in webserver
-                # due to XComs mixed with pickled and unpickled.
-                self.value = pickle.loads(self.value)
 
     def __repr__(self):
         return '<XCom "{key}" ({task_id} @ {execution_date})>'.format(
@@ -233,6 +229,23 @@ def serialize_value(value):
                       "support for XCOM in your airflow config.")
             raise
 
+    @staticmethod
+    def deserialize_value(result):
+        # TODO: "pickling" has been deprecated and JSON is preferred.
+        # "pickling" will be removed in Airflow 2.0.
+        enable_pickling = conf.getboolean('core', 'enable_xcom_pickling')
+        if enable_pickling:
+            return pickle.loads(result.value)
+
+        try:
+            return json.loads(result.value.decode('UTF-8'))
+        except ValueError:
+            log.error("Could not deserialize the XCOM value from JSON. "
+                      "If you are using pickles instead of JSON "
+                      "for XCOM, then you need to enable pickle "
+                      "support for XCOM in your airflow config.")
+            raise
+
 
 def resolve_xcom_backend():
     """Resolves custom XCom class"""