fire GenericExceptionOnRun for batch-level exception

core/dbt/task/run.py

@@ -27,11 +27,11 @@
 from dbt.contracts.graph.manifest import Manifest
 from dbt.contracts.graph.nodes import HookNode, ModelNode, ResultNode
 from dbt.events.types import (
+    GenericExceptionOnRun,
     LogHookEndLine,
     LogHookStartLine,
     LogModelResult,
     LogStartLine,
-    RunningOperationCaughtError,
 )
 from dbt.exceptions import CompilationError, DbtInternalError, DbtRuntimeError
 from dbt.graph import ResourceTypeSelector
@@ -275,7 +275,13 @@ def print_batch_result_line(
             level=level,
         )
         if exception:
-            fire_event(RunningOperationCaughtError(exc=str(exception)))
+            fire_event(
+                GenericExceptionOnRun(
+                    unique_id=self.node.unique_id,
+                    exc=f"Exception on worker thread. {str(exception)}",
+                    node_info=self.node.node_info,
+                )
+            )
 
     def print_batch_start_line(
         self, batch_start: Optional[datetime], batch_idx: int, batch_total: int

tests/functional/microbatch/test_microbatch.py

@@ -3,6 +3,7 @@
 import pytest
 
 from dbt.events.types import (
+    GenericExceptionOnRun,
     LogModelResult,
     MicrobatchMacroOutsideOfBatchesDeprecation,
     MicrobatchModelNoEventTimeInputs,
@@ -526,9 +527,15 @@ def models(self):
         }
 
     def test_run_with_event_time(self, project):
-        # run all partitions from start - 2 expected rows in output, one failed
+        event_catcher = EventCatcher(
+            GenericExceptionOnRun, predicate=lambda event: event.data.node_info is not None
+        )
+
         with patch_microbatch_end_time("2020-01-03 13:57:00"):
-            run_dbt(["run"], expect_pass=False)
+            run_dbt(["run"], callbacks=[event_catcher.catch], expect_pass=False)
+
+        assert len(event_catcher.caught_events) == 1
+        # run all partitions from start - 2 expected rows in output, one failed
         self.assert_row_count(project, "microbatch_model", 2)
 
         run_results = get_artifact(project.project_root, "target", "run_results.json")