title: Error : SQL insertion error at the time of task failed using CloudSqlQueryOperator
body: <p>I have some doubt related to my python code. Here I have created one DAG with two task one is executing fine and another one is failed(forcefully I have mentioned wrong values for check the scenario).</p>
  
  <p>I am trying to implement on_failure_callback at the time of task failure. and at the time of task failure need to insert the error message in GCP mysql. </p>
  
  <p>Here in below code I have one method "sql_failed_task" which is responsible for pushing the error message and calling CloudSqlQueryOperator. </p>
  
  <p>But Insertion is not working properly : Executing: "["INSERT INTO task_exception (failure_reason) VALUES ('{{ti.xcom_pull(key='the_message')}}')"]" </p>
  
  <p>error : You have an error in your SQL syntax; check the manual that corresponds to your MySQL server version for the right syntax to use near 'the_message')}}')' at line 1")</p>
  
  <p>Below code(There is no indention error in my code, might be you can see here because of some editing or copy paste) : </p>
  
  <pre><code>import os
  
  from airflow import DAG
  from airflow.operators.python_operator import PythonOperator
  from airflow.operators.bash_operator import BashOperator
  from airflow.utils.trigger_rule import TriggerRule
  from datetime import datetime, timedelta
  from airflow.contrib.operators import dataproc_operator
  from airflow import models
  from airflow.hooks.mysql_hook import MySqlHook
  from urllib.parse import quote_plus
  from airflow.models import Variable
  from airflow.contrib.operators.gcp_sql_operator import 
  CloudSqlQueryOperator
  
  db_config = Variable.get("analytics_audit_db_properties", 
  deserialize_json=True)
  
  GCSQL_MYSQL_USER =db_config["GCSQL_MYSQL_USER"]
  GCSQL_MYSQL_PASSWORD =db_config["GCSQL_MYSQL_PASSWORD"]
  GCSQL_MYSQL_PUBLIC_PORT =db_config["GCSQL_MYSQL_PUBLIC_PORT"]
  GCSQL_MYSQL_DATABASE_NAME =db_config["GCSQL_MYSQL_DATABASE_NAME"]
  GCP_PROJECT_ID=db_config["GCP_PROJECT_ID"]
  GCP_REGION=db_config["GCP_REGION"]
  GCSQL_MYSQL_INSTANCE_NAME_QUERY 
  =db_config["GCSQL_MYSQL_INSTANCE_NAME_QUERY"]
  
  SQL = [
       "INSERT INTO task_exception (failure_reason) VALUES 
  ('{{ti.xcom_pull(key='the_message')}}')",
  ]
  
   mysql_kwargs = dict(
      user=quote_plus(GCSQL_MYSQL_USER),
      password=quote_plus(GCSQL_MYSQL_PASSWORD),
      public_port=GCSQL_MYSQL_PUBLIC_PORT,
      project_id=quote_plus(GCP_PROJECT_ID),
      location=quote_plus(GCP_REGION),
      instance=quote_plus(GCSQL_MYSQL_INSTANCE_NAME_QUERY),
      database=quote_plus(GCSQL_MYSQL_DATABASE_NAME)
  )
  
  os.environ['AIRFLOW_CONN_PROXY_MYSQL_TCP'] = \
      "gcpcloudsql://{user}:{password}@{public_port}/{database}?" \
      "database_type=mysql&amp;" \
      "project_id={project_id}&amp;" \
      "location={location}&amp;" \
      "instance={instance}&amp;" \
      "use_proxy=True&amp;" \
      "sql_proxy_use_tcp=True".format(**mysql_kwargs)
  
  connection_names = [
      "proxy_mysql_tcp",
  ]
  tasks = []
  
  default_args = {
      'owner': 'airflow',
      'depends_on_past': False,
      'start_date': datetime(2019, 1, 1),
      'email_on_failure': True,
      'email_on_retry': False,
      'retries': 0,
      'retry_delay': timedelta(minutes=30),    
  }
  
  with models.DAG(
          dag_id='Dag_failure_callback_example_2',
          default_args=default_args,
          schedule_interval=timedelta(days=1)
   ) as dag:
  
  def my_task_py():
          print('Hello World')
  
  def sql_failed_task(context):
      print('inside sql_failed_task')
      print("context exception : ", context.get('exception'))
      task_instance = context['task_instance']
      exception = context.get('exception')
      task_instance.xcom_push(key="the_message", value=str(exception))
      print('after push')
      connection_name = 'proxy_mysql_tcp'
      failed_alert = CloudSqlQueryOperator(
          gcp_cloudsql_conn_id=connection_name,
          task_id="sql_failed_task" + connection_name,
          sql=SQL)
      failed_alert.execute(context=context)
  
  
  
  
      my_task_1 = PythonOperator(task_id='my_task_1', 
  python_callable=my_task_py, dag=dag)
  
  
  create_dataproc_cluster = dataproc_operator.DataprocClusterCreateOperator(
      task_id='create_dataproc_cluster',
      project_id='trim********',
      cluster_name='airflow-*****',
      num_workers=2,
      zone='europe-west1-c',
      master_machine_type='n1-standard-1',
      worker_machine_type='n1-standard-1',
      on_failure_callback=sql_failed_task,
      dag=dag)
  
      my_task_1 &gt;&gt; create_dataproc_cluster
  </code></pre>