set ErrorSource when retry request limit exceeded (#6806)

resolves #6219 ## Test Plan ![Screen Shot 2022-02-25 at 9.50.28 AM.png](https://graphite-user-uploaded-assets.s3.amazonaws.com/SYNwsA1kHFPT6jksBdKE/cef5b226-f11d-4e56-9d1e-925b47eddf1d/Screen%20Shot%202022-02-25%20at%209.50.28%20AM.png) screenshots attaching
dagster-io · Feb 28, 2022 · ca94d8d · ca94d8d
1 parent c157260
commit ca94d8d
Showing 1 changed file with 7 additions and 2 deletions.
diff --git a/python_modules/dagster/dagster/core/execution/plan/execute_plan.py b/python_modules/dagster/dagster/core/execution/plan/execute_plan.py
@@ -242,15 +242,20 @@ def _dagster_event_sequence_for_step(step_context: StepExecutionContext) -> Iter
             prev_attempts = step_context.previous_attempt_count
             if prev_attempts >= retry_request.max_retries:
                 fail_err = SerializableErrorInfo(
-                    message="Exceeded max_retries of {}".format(retry_request.max_retries),
+                    message=f"Exceeded max_retries of {retry_request.max_retries}\n",
                     stack=retry_err_info.stack,
                     cls_name=retry_err_info.cls_name,
                     cause=retry_err_info.cause,
                 )
                 step_context.capture_step_exception(retry_request)
                 yield DagsterEvent.step_failure_event(
                     step_context=step_context,
-                    step_failure_data=StepFailureData(error=fail_err, user_failure_data=None),
+                    step_failure_data=StepFailureData(
+                        error=fail_err,
+                        user_failure_data=None,
+                        # set the flag to omit the outer stack if we have a cause to show
+                        error_source=ErrorSource.USER_CODE_ERROR if fail_err.cause else None,
+                    ),
                 )
             else:
                 yield DagsterEvent.step_retry_event(