<h3 dir="auto">Description</h3>
<p dir="auto">Given a simple dag like this:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from airflow import DAG
from airflow.utils.dates import days_ago
from airflow.operators.bash import BashOperator

with DAG(&quot;instantfail&quot;,
         description=&quot;instantfail&quot;,
         tags=[&quot;example&quot;],
         catchup=False,
         schedule_interval=None,
         start_date=days_ago(2),
        ) as dag:

    BashOperator(bash_command=&quot;echo hello &amp;&amp; false&quot;, task_id=&quot;b&quot;)"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">airflow</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">utils</span>.<span class="pl-s1">dates</span> <span class="pl-k">import</span> <span class="pl-s1">days_ago</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">bash</span> <span class="pl-k">import</span> <span class="pl-v">BashOperator</span>

<span class="pl-k">with</span> <span class="pl-v">DAG</span>(<span class="pl-s">"instantfail"</span>,
         <span class="pl-s1">description</span><span class="pl-c1">=</span><span class="pl-s">"instantfail"</span>,
         <span class="pl-s1">tags</span><span class="pl-c1">=</span>[<span class="pl-s">"example"</span>],
         <span class="pl-s1">catchup</span><span class="pl-c1">=</span><span class="pl-c1">False</span>,
         <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>,
         <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-en">days_ago</span>(<span class="pl-c1">2</span>),
        ) <span class="pl-k">as</span> <span class="pl-s1">dag</span>:

    <span class="pl-v">BashOperator</span>(<span class="pl-s1">bash_command</span><span class="pl-c1">=</span><span class="pl-s">"echo hello &amp;&amp; false"</span>, <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"b"</span>)</pre></div>
<p dir="auto">The error log from the failing bash command, <code class="notranslate">false</code>, becomes incredibly big with lots of irrelevant information for the writer of the DAG, if that's not enough, the error is printed twice:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="[2021-12-02, 17:07:01 UTC] {subprocess.py:85} INFO - Output:
[2021-12-02, 17:07:01 UTC] {subprocess.py:89} INFO - hello
[2021-12-02, 17:07:01 UTC] {subprocess.py:93} INFO - Command exited with return code 1
[2021-12-02, 17:07:01 UTC] {taskinstance.py:1703} ERROR - Task failed with exception
Traceback (most recent call last):
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py&quot;, line 1332, in _run_raw_task
    self._execute_task_with_callbacks(context)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py&quot;, line 1458, in _execute_task_with_callbacks
    result = self._execute_task(context, self.task)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py&quot;, line 1514, in _execute_task
    result = execute_callable(context=context)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/operators/bash.py&quot;, line 187, in execute
    raise AirflowException(
airflow.exceptions.AirflowException: Bash command failed. The command returned a non-zero exit code 1.
[2021-12-02, 17:07:01 UTC] {taskinstance.py:1270} INFO - Marking task as FAILED. dag_id=instantfail, task_id=b, execution_date=20211202T160659, start_date=20211202T160701, end_date=20211202T160701
[2021-12-02, 17:07:01 UTC] {standard_task_runner.py:88} ERROR - Failed to execute job 292 for task b
Traceback (most recent call last):
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/task/task_runner/standard_task_runner.py&quot;, line 85, in _start_by_fork
    args.func(args, dag=self.dag)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/cli/cli_parser.py&quot;, line 48, in command
    return func(*args, **kwargs)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/utils/cli.py&quot;, line 92, in wrapper
    return f(*args, **kwargs)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/cli/commands/task_command.py&quot;, line 292, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/cli/commands/task_command.py&quot;, line 107, in _run_task_by_selected_method
    _run_raw_task(args, ti)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/cli/commands/task_command.py&quot;, line 180, in _run_raw_task
    ti._run_raw_task(
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/utils/session.py&quot;, line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py&quot;, line 1332, in _run_raw_task
    self._execute_task_with_callbacks(context)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py&quot;, line 1458, in _execute_task_with_callbacks
    result = self._execute_task(context, self.task)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py&quot;, line 1514, in _execute_task
    result = execute_callable(context=context)
  File &quot;/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/operators/bash.py&quot;, line 187, in execute
    raise AirflowException(
airflow.exceptions.AirflowException: Bash command failed. The command returned a non-zero exit code 1.
[2021-12-02, 17:07:01 UTC] {local_task_job.py:154} INFO - Task exited with return code 1
[2021-12-02, 17:07:01 UTC] {local_task_job.py:264} INFO - 0 downstream tasks scheduled from follow-on schedule check"><pre class="notranslate"><code class="notranslate">[2021-12-02, 17:07:01 UTC] {subprocess.py:85} INFO - Output:
[2021-12-02, 17:07:01 UTC] {subprocess.py:89} INFO - hello
[2021-12-02, 17:07:01 UTC] {subprocess.py:93} INFO - Command exited with return code 1
[2021-12-02, 17:07:01 UTC] {taskinstance.py:1703} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1332, in _run_raw_task
    self._execute_task_with_callbacks(context)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1458, in _execute_task_with_callbacks
    result = self._execute_task(context, self.task)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1514, in _execute_task
    result = execute_callable(context=context)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/operators/bash.py", line 187, in execute
    raise AirflowException(
airflow.exceptions.AirflowException: Bash command failed. The command returned a non-zero exit code 1.
[2021-12-02, 17:07:01 UTC] {taskinstance.py:1270} INFO - Marking task as FAILED. dag_id=instantfail, task_id=b, execution_date=20211202T160659, start_date=20211202T160701, end_date=20211202T160701
[2021-12-02, 17:07:01 UTC] {standard_task_runner.py:88} ERROR - Failed to execute job 292 for task b
Traceback (most recent call last):
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/task/task_runner/standard_task_runner.py", line 85, in _start_by_fork
    args.func(args, dag=self.dag)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/utils/cli.py", line 92, in wrapper
    return f(*args, **kwargs)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 292, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 107, in _run_task_by_selected_method
    _run_raw_task(args, ti)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 180, in _run_raw_task
    ti._run_raw_task(
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1332, in _run_raw_task
    self._execute_task_with_callbacks(context)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1458, in _execute_task_with_callbacks
    result = self._execute_task(context, self.task)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1514, in _execute_task
    result = execute_callable(context=context)
  File "/homedir/.local/share/virtualenvs/airflow-uWHa5xnK/lib/python3.9/site-packages/airflow/operators/bash.py", line 187, in execute
    raise AirflowException(
airflow.exceptions.AirflowException: Bash command failed. The command returned a non-zero exit code 1.
[2021-12-02, 17:07:01 UTC] {local_task_job.py:154} INFO - Task exited with return code 1
[2021-12-02, 17:07:01 UTC] {local_task_job.py:264} INFO - 0 downstream tasks scheduled from follow-on schedule check
</code></pre></div>
<p dir="auto">Often operators can be be more advanced, especially PythonOperator and taskflow which frequently creates equally deep stack traces themselves, which is usually what you are interested in, but together it accumulates into several pages of error.</p>
<p dir="auto">I'm usually the guy who thinks a stack trace is the best error message and like the context it provides, but here i think it goes a bit too far.</p>
<hr>
<ol dir="auto">
<li>Why is the error printed twice?</li>
<li>The writer of a dag should be able to assume the execution path up to execute_callable() was sane. If i'm debugging a DAG or an Operator i don't want to debug Airflow task runner internals. I don't know how true this is but this should be the goal at least.</li>
</ol>
<p dir="auto">Expected log output should only contain</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="  File &quot;airflow/operators/bash.py&quot;, line 187, in execute
    raise AirflowException(
airflow.exceptions.AirflowException: Bash command failed. The command returned a non-zero exit code 1."><pre class="notranslate"><code class="notranslate">  File "airflow/operators/bash.py", line 187, in execute
    raise AirflowException(
airflow.exceptions.AirflowException: Bash command failed. The command returned a non-zero exit code 1.
</code></pre></div>
<p dir="auto">Can <code class="notranslate">TaskInstance._execute_task</code> be made to catch the operators exception and rethrow as something else, that can be ignored at the end by <code class="notranslate">standard_task_runner</code>?</p>
<p dir="auto">Or is this something that simply should be solved by logging configuration?</p>
<h3 dir="auto">Use case/motivation</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Related issues</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit a PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.1.4</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">At my organization, we currently followed the documentation for Bigquery data transfer in Airflow/Composer (1.17.9). It calls a <code class="notranslate">BigQueryDataTransferServiceTransferRunSensor</code>  which executes and creates a class called <code class="notranslate">BigQueryDataTransferServiceHook</code>.  Using this hook object, it calls a specific function called <code class="notranslate">get_transfer_run</code> which calls its counterpart in grpc.</p>
<p dir="auto">Using the passed arguments, it creates the request name using f-strings which is passed into the function to fetch for the transfer config object.</p>
<p dir="auto">In the <a href="https://github.com/apache/airflow/blob/main/airflow/providers/google/cloud/hooks/bigquery_dts.py">bigquery_dts file</a>, at line 263 using <code class="notranslate">providers-google@6.3.0</code> or main branch, it doesn't interpolate the variables to the string, <code class="notranslate">name</code>, correctly due to a misplaced f.</p>
<p dir="auto">This results in the value, <code class="notranslate">"f{project}/transferConfigs/{transfer_config_id}/runs/{run_id}"</code>, for <code class="notranslate">name</code> which causes an "invalid run name" error from grpc.</p>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">Let's say I have these variables containing the data transfer run I want to monitor passed into the Sensor class:</p>
<ul dir="auto">
<li>project_id = "123"</li>
<li>transfer_config_id = "456"</li>
<li>run_id = "789"</li>
</ul>
<p dir="auto">I expect the variable name to contain <code class="notranslate">projects/123/transferConfigs/456/runs/789</code> when calling <code class="notranslate">get_transfer_run</code> with the above arguments during execution.</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">This is assuming you have a GCP project and a dataset with any kind of tables in it.</p>
<ol dir="auto">
<li>Go to Bigquery Data Transfer which is on the sidebar of Bigquery workspace or search "Bigquery Data Transfer".</li>
<li>Click on "CREATE TRANSFER"</li>
<li>Select "Dataset Copy"</li>
<li>Fill in all the required information and set "Repeats" from "daily" to "On-demand".</li>
<li>Click on "SAVE" and it should bring you back into its homepage with the created transfer config.</li>
<li>Click on the created transfer config and click "RUN TRANSFER NOW". This should generate an "in-progress" run.</li>
<li>Click on the completed run and it should bring up a sidebar from the right containing the "Run details".</li>
<li>In <code class="notranslate">Resource name</code>, save the numbers to the right of <code class="notranslate">Projects</code>, <code class="notranslate">transferConfigs</code> and <code class="notranslate">runs</code> as they are be the information needed to run the Sensor class in Airflow.</li>
<li>From Airflow, a basic dag file with <code class="notranslate">BigQueryDataTransferServiceTransferRunSensor</code> with the three information from above.</li>
<li>Execute the dag and it will produce an error with an invalid run name</li>
</ol>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu 18.04.06 LTS</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow-providers-google==6.3.0</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Composer</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto">None just default Composer</p>
<h3 dir="auto">Anything else</h3>
<p dir="auto">I would be down to create a PR for this as it seems like a chill bug and would be a great exposure for me being new to all this. However, I'm fine leaving this for others to do if it's a lot more dire than expected.</p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>