async callback exceptions were not handled properly (no capturing up the stack)

pypushflow/PythonActor.py

@@ -118,7 +118,7 @@ class PythonActor(AbstractActor):
         self.listErrorHandler.append(errorHandler)
 
     def trigger(self, inData: dict):
-        logger.info("In trigger %s, inData = %s", self.name, pprint.pformat(inData))
+        logger.debug("In trigger %s, inData = %s", self.name, pprint.pformat(inData))
         self.setStarted()
         self.inData = dict(inData)
         self.uploadInDataToMongo(actorData={"inData": inData}, script=self.script)
@@ -141,29 +141,38 @@ class PythonActor(AbstractActor):
             self.af.call(**self.inData)
 
     def resultHandler(self, result: dict):
-        # Handle the result
-        logger.debug("In resultHandler for '%s'", self.name)
-        self._finishedSuccess(result)
-
-        # Trigger actors
-        downstreamData = dict(self.inData)
-        downstreamData.update(result)
-        self._triggerDownStreamActors(downstreamData)
+        """Async callback in case of success"""
+        try:
+            # Handle the result
+            logger.debug("In resultHandler for '%s'", self.name)
+            self._finishedSuccess(result)
+
+            # Trigger actors
+            downstreamData = dict(self.inData)
+            downstreamData.update(result)
+            self._triggerDownStreamActors(downstreamData)
+        except Exception as e:
+            self.errorHandler(e)
 
     def errorHandler(self, exception: Exception):
-        # Handle the result
-        logger.error(
-            "Error in python actor '%s'! Not running down stream actors %s",
-            self.name,
-            [actor.name for actor in self.listDownStreamActor],
-        )
-        result = self._parseException(exception)
-        self._finishedFailure(result)
-
-        # Trigger actors
-        downstreamData = dict(self.inData)
-        downstreamData["WorkflowException"] = result
-        self._triggerErrorHandlers(downstreamData)
+        """Async callback in case of exception"""
+        try:
+            # Handle the result
+            logger.debug("In errorHandler for '%s'", self.name)
+            logger.error(
+                "Error in python actor '%s'! Not running down stream actors %s",
+                self.name,
+                [actor.name for actor in self.listDownStreamActor],
+            )
+            result = self._parseException(exception)
+            self._finishedFailure(result)
+
+            # Trigger actors
+            downstreamData = dict(self.inData)
+            downstreamData["WorkflowException"] = result
+            self._triggerErrorHandlers(downstreamData)
+        except Exception:
+            logger.exception("In errorHandler for '%s'", self.name)
 
     def _parseException(self, exception: Exception) -> dict:
         errorMessage = str(exception)