sqlalchemy.exc.OperationalError during dag-processor

[asosnovsky-sumologic]

Apache Airflow version

2.5.3

What happened

While our dag-processor it keeps coming across this error

0s           2023-04-03T14:30:22
================================================================================
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1880, in _execute_context
    self.dialect.do_executemany(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/dialects/mysql/mysqldb.py", line 193, in do_executemany
    rowcount = cursor.executemany(statement, parameters)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/cursors.py", line 239, in executemany
    self.rowcount = sum(self.execute(query, arg) for arg in args)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/cursors.py", line 239, in <genexpr>
    self.rowcount = sum(self.execute(query, arg) for arg in args)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/cursors.py", line 206, in execute
    res = self._query(query)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/cursors.py", line 319, in _query
    db.query(q)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/connections.py", line 254, in query
    _mysql.connection.query(self, query)
MySQLdb.OperationalError: (2013, 'Lost connection to MySQL server during query')

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/home/airflow/.local/bin/airflow", line 8, in <module>
    sys.exit(main())
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/__main__.py", line 39, in main
    args.func(args)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/cli/cli_parser.py", line 52, in command
    return func(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/cli.py", line 108, in wrapper
    return f(*args, **kwargs)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/cli/commands/dag_processor_command.py", line 83, in dag_processor
    manager.start()
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 489, in start
    return self._run_parsing_loop()
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 601, in _run_parsing_loop
    self._fetch_callbacks(max_callbacks_per_loop)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/session.py", line 75, in wrapper
    return func(*args, session=session, **kwargs)
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/dag_processing/manager.py", line 687, in _fetch_callbacks
    guard.commit()
  File "/home/airflow/.local/lib/python3.10/site-packages/airflow/utils/sqlalchemy.py", line 461, in commit
    self.session.commit()
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 1451, in commit
    self._transaction.commit(_to_root=self.future)
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 829, in commit
    self._prepare_impl()
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 808, in _prepare_impl
    self.session.flush()
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3444, in flush
    self._flush(objects)
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3583, in _flush
    with util.safe_reraise():
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/langhelpers.py", line 70, in __exit__
    compat.raise_(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/session.py", line 3544, in _flush
    flush_context.execute()
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/unitofwork.py", line 456, in execute
    rec.execute(self)
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/unitofwork.py", line 667, in execute
    util.preloaded.orm_persistence.delete_obj(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/persistence.py", line 343, in delete_obj
    _emit_delete_statements(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/orm/persistence.py", line 1468, in _emit_delete_statements
    c = connection._execute_20(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1705, in _execute_20
    return meth(self, args_10style, kwargs_10style, execution_options)
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/sql/elements.py", line 334, in _execute_on_connection
    return connection._execute_clauseelement(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1572, in _execute_clauseelement
    ret = self._execute_context(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1943, in _execute_context
    self._handle_dbapi_exception(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 2124, in _handle_dbapi_exception
    util.raise_(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/util/compat.py", line 211, in raise_
    raise exception
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/engine/base.py", line 1880, in _execute_context
    self.dialect.do_executemany(
  File "/home/airflow/.local/lib/python3.10/site-packages/sqlalchemy/dialects/mysql/mysqldb.py", line 193, in do_executemany
    rowcount = cursor.executemany(statement, parameters)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/cursors.py", line 239, in executemany
    self.rowcount = sum(self.execute(query, arg) for arg in args)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/cursors.py", line 239, in <genexpr>
    self.rowcount = sum(self.execute(query, arg) for arg in args)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/cursors.py", line 206, in execute
    res = self._query(query)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/cursors.py", line 319, in _query
    db.query(q)
  File "/home/airflow/.local/lib/python3.10/site-packages/MySQLdb/connections.py", line 254, in query
    _mysql.connection.query(self, query)
sqlalchemy.exc.OperationalError: (MySQLdb.OperationalError) (2013, 'Lost connection to MySQL server during query')
[SQL: DELETE FROM callback_request WHERE callback_request.id = %s]
[parameters: ((25156,), (25157,), (25158,), (25159,), (25160,), (25161,), (25162,), (25163,)  ... displaying 10 of 20 total bound parameter sets ...  (25175,), (25176,))]

What you think should happen instead

dag-processor should complete without an error

How to reproduce

Run airflow pod with dag-processor, wait 10 minutes

We run our airflow based on the https://github.com/airflow-helm/charts chart, we've forked one of their deployments like this:

{{ if .Values.customEnv.flags.enableDagProcessor }}
{{- $airflowValues := ( dict "Values" .Values.airflow "Release" .Release) -}}
{{- $podNodeSelector := include "airflow.podNodeSelector" (dict "Release" .Release "Values" .Values.airflow "nodeSelector" .Values.airflow.scheduler.nodeSelector) }}
{{- $podAffinity := include "airflow.podAffinity" (dict "Release" .Release "Values" .Values.airflow "affinity" .Values.airflow.scheduler.affinity) }}
{{- $podTolerations := include "airflow.podTolerations" (dict "Release" .Release "Values" .Values.airflow "tolerations" .Values.airflow.scheduler.tolerations) }}
{{- $podSecurityContext := include "airflow.podSecurityContext" (dict "Release" .Release "Values" .Values.airflow "securityContext" .Values.airflow.scheduler.securityContext) }}
{{- $extraPipPackages := concat .Values.airflow.airflow.extraPipPackages .Values.airflow.scheduler.extraPipPackages }}
{{- $extraVolumeMounts := .Values.airflow.scheduler.extraVolumeMounts }}
{{- $volumeMounts := include "airflow.volumeMounts" (dict "Release" .Release "Values" .Values.airflow "extraPipPackages" $extraPipPackages "extraVolumeMounts" $extraVolumeMounts) }}
{{- $extraVolumes := .Values.airflow.scheduler.extraVolumes }}
{{- $volumes := include "airflow.volumes" (dict "Release" .Release "Values" .Values.airflow "extraPipPackages" $extraPipPackages "extraVolumes" $extraVolumes "extraVolumeMounts" $extraVolumeMounts) }}
apiVersion: apps/v1
kind: Deployment
metadata:
  name: {{ include "airflow.fullname" . }}-dag-processor
  labels:
    app: {{ include "airflow.labels.app" . }}
    component: dag-processor
    chart: {{ include "airflow.labels.chart" . }}
    release: {{ .Release.Name }}
    heritage: {{ .Release.Service }}
spec:
  replicas: 1
  strategy:
    type: Recreate
  selector:
    matchLabels:
      app: {{ include "airflow.labels.app" . }}
      component: dag-processor
      release: {{ .Release.Name }}
  template:
    metadata:
      annotations:
        {{- if .Values.airflow.airflow.podAnnotations }}
        {{- toYaml .Values.airflow.airflow.podAnnotations | nindent 8 }}
        {{- end }}
      labels:
        app: {{ include "airflow.labels.app" . }}
        component: dag-processor
        release: {{ .Release.Name }}
    spec:
      restartPolicy: Always
      {{- if .Values.airflow.airflow.image.pullSecret }}
      imagePullSecrets:
        - name: {{ .Values.airflow.airflow.image.pullSecret }}
      {{- end }}
      {{- if $podNodeSelector }}
      nodeSelector:
        {{- $podNodeSelector | nindent 8 }}
      {{- end }}
      {{- if $podAffinity }}
      affinity:
        {{- $podAffinity | nindent 8 }}
      {{- end }}
      {{- if $podTolerations }}
      tolerations:
        {{- $podTolerations | nindent 8 }}
      {{- end }}
      {{- if $podSecurityContext }}
      securityContext:
        {{- $podSecurityContext | nindent 8 }}
      {{- end }}
      serviceAccountName: {{ include "airflow.serviceAccountName" . }}
      initContainers:
        {{- if $extraPipPackages }}
        {{- include "airflow.init_container.install_pip_packages" (dict "Release" .Release "Values" .Values.airflow "extraPipPackages" $extraPipPackages) | indent 8 }}
        {{- end }}
        {{- if .Values.airflow.dags.gitSync.enabled }}
        {{- include "airflow.container.git_sync" (dict "Release" .Release "Values" .Values.airflow "sync_one_time" "true") | indent 8 }}
        {{- end }}
        {{- include "airflow.init_container.check_db" (dict "Release" .Release "Values" .Values.airflow "volumeMounts" $volumeMounts) | indent 8 }}
        {{- include "airflow.init_container.wait_for_db_migrations" (dict "Release" .Release "Values" .Values.airflow "volumeMounts" $volumeMounts) | indent 8 }}
      containers:
        - name: airflow-dag-processor
          {{- include "airflow.image" $airflowValues | indent 10 }}
          resources: 
            limits:
              cpu: 1000m
              memory: 1Gi
            requests:
              cpu: 600m
              memory: 500Mi
          envFrom:
            {{- include "airflow.envFrom" . | indent 12 }}
          env:
            {{- include "airflow.env" $airflowValues | indent 12 }}
          command:
            {{- include "airflow.command" $airflowValues | indent 12 }}
          args:
            - "bash"
            - "-c"
            - "exec airflow dag-processor"
          {{- if or ($volumeMounts) (include "airflow.executor.kubernetes_like" $airflowValues) }}
          volumeMounts:
            {{- $volumeMounts | indent 12 }}
            {{- if include "airflow.executor.kubernetes_like" $airflowValues }}
            - name: pod-template
              mountPath: /opt/airflow/pod_templates/pod_template.yaml
              subPath: pod_template.yaml
              readOnly: true
            {{- end }}
          {{- end }}
        {{- include "airflow.container.git_sync" $airflowValues | indent 8 }}
        {{- if .Values.airflow.airflow.extraContainers }}
        {{- toYaml .Values.airflow.airflow.extraContainers | nindent 8 }}
        {{- end }}
      {{- if or ($volumes) (include "airflow.executor.kubernetes_like" $airflowValues) }}
      volumes:
        {{- $volumes | indent 8 }}
        {{- if include "airflow.executor.kubernetes_like" $airflowValues }}
        - name: pod-template
          configMap:
            name: {{ include "airflow.fullname" . }}-pod-template
        {{- end }}
      {{- end }}
{{ end }}

Operating System

PRETTY_NAME="Debian GNU/Linux 11 (bullseye)" NAME="Debian GNU/Linux" VERSION_ID="11" VERSION="11 (bullseye)" VERSION_CODENAME=bullseye ID=debian HOME_URL="https://www.debian.org/" SUPPORT_URL="https://www.debian.org/support" BUG_REPORT_URL="https://bugs.debian.org/"

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==7.1.0
apache-airflow-providers-celery==3.1.0
apache-airflow-providers-cncf-kubernetes==5.1.1
apache-airflow-providers-common-sql==1.3.3
apache-airflow-providers-docker==3.4.0
apache-airflow-providers-elasticsearch==4.3.3
apache-airflow-providers-ftp==3.3.0
apache-airflow-providers-google==8.8.0
apache-airflow-providers-grpc==3.1.0
apache-airflow-providers-hashicorp==3.2.0
apache-airflow-providers-http==4.1.1
apache-airflow-providers-imap==3.1.1
apache-airflow-providers-microsoft-azure==5.1.0
apache-airflow-providers-mysql==4.0.0
apache-airflow-providers-odbc==3.2.1
apache-airflow-providers-postgres==5.4.0
apache-airflow-providers-redis==3.1.0
apache-airflow-providers-sendgrid==3.1.0
apache-airflow-providers-sftp==4.2.1
apache-airflow-providers-slack==7.2.0
apache-airflow-providers-sqlite==3.3.1
apache-airflow-providers-ssh==3.4.0

Deployment

Other 3rd-party Helm chart

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[potiuk]

You need to see why your server keeps on closing the connection. This looks like instability of your mysql server. It could have been caused by some misconfiguration of your serve, but only your sever logs can tell it. Please take a look there.

Converting into a discussion - maybe looking at your server logs will shed some more light.

[asosnovsky-sumologic]

well here are the logs that we see when this happens

2023-04-10T14:20:11.545167Z 0 [Note] [MY-011880] [InnoDB] Un-latching whole of buffer pool. (buf0buf.cc:2575)
--
2023-04-10T14:20:11.546826Z 0 [Note] [MY-011894] [InnoDB] Completed to resize buffer pool from 2415919104 to 3221225472. (buf0buf.cc:3804)
2023-04-10T14:20:11.546861Z 0 [Note] [MY-011880] [InnoDB] Completed resizing buffer pool at 230410 14:20:11. (buf0buf.cc:2575)
2023-04-10T14:20:13.749313Z 2867493 [Note] [MY-010914] [Server] Aborted connection 2867493 to db: 'airflow' user: 'admin' host: 'x.x.x.x.x' (Got an error reading communication packets). (sql_connect.cc:835)
2023-04-10T14:20:15.957595Z 2868300 [Note] [MY-010914] [Server] Aborted connection 2868300 to db: 'airflow' user: 'admin' host: 'x.x.x.x.x' (Got an error reading communication packets). (sql_connect.cc:835)
2023-04-10T14:20:25.868888Z 0 [Note] [MY-011880] [InnoDB] No size change: 3221225472. Nothing to do. (buf0buf.cc:2575)

please note we are using mysql with AWS serverless aurora, and here is what I found in their docs about this error
https://repost.aws/knowledge-center/rds-mysql-communication-packet-error

based on this we've tried to set the parameter max_allowed_packet since the failure seems to happen when a big query against callback_request is occuring.

This fix was deployed today, and so far it seems to have resolved our issue. I will monitor it going forward and comment here if we see more failures. If you have any other ideas on what to do please let us know :)

[asosnovsky-sumologic]

I can confirm that after 1 hour the same error remains, with the same logs

[asosnovsky-sumologic]

we've set the value to it's maximum,
additionally we've increased pretty much each of the timeout variables to up 10min and are still experiencing the issue

[asosnovsky-sumologic]

So the following parameter change seems to have calmed things down...

+ parameter {
+   apply_method = "immediate"
+   name         = "interactive_timeout"
+   value        = "60000"
}
+ parameter {
+   apply_method = "immediate"
+   name         = "max_allowed_packet"
+   value        = "1073741824"
}
+ parameter {
+   apply_method = "immediate"
+   name         = "net_read_timeout"
+   value        = "60000"
}
+ parameter {
+   apply_method = "immediate"
+   name         = "net_write_timeout"
+   value        = "60000"
}
+ parameter {
+   apply_method = "immediate"
+   name         = "wait_timeout"
+   value        = "60000"
}

but I've basically set them to their maximum value. I feel like something is unstable in terms of how airflow is handling connections when using the dag-processor

[potiuk]

cc: @mhenc - might be good idea to take a look if there is something special around standalone dag file processor that could generate it?

[lzf12]

Our connection is also often disconnected

[potiuk]

This means you should fix the instabilities of your system - either by looking of your networking or database configuration (for example dropping the connnections because it uses too much memory or similar) - you should look at your networking configuration and logs and your database logs. There is not much airflow can do if the DB connection is dropped in-the-middle. On the other you, or whoever is a Deployment Manager for your installatoin (maybe even when they run it as a service) are the right one to look at those issues.