Elasticsearch provider 6.5.1 (Airflow 3.2.1) sends compatible-with=9 header to ES 8.x server, breaking remote task log ingestion

[petercheon]

Apache Airflow version

3.2.1

Apache Airflow Provider versions

apache-airflow-providers-elasticsearch == 6.5.1

What happened

After upgrading from Airflow 3.2.0 to 3.2.1 (which bundles apache-airflow-providers-elasticsearch 6.5.1), task log ingestion to Elasticsearch 8.x servers stops working. Every _bulk request from the worker is rejected by the ES server with HTTP 400 media_type_header_exception, because the elasticsearch-py client now sends Accept: application/vnd.elasticsearch+json; compatible-with=9 while the ES 8.x server only accepts compatible-with=7 or 8.

The change appears to come from #64070 ("Support Elasticsearch 9 in Elasticsearch Provider"), which bumped the dependency to elasticsearch>=8.10,<10. With the ES 9 client installed against an ES 8 server, the client unconditionally negotiates compatible-with=9, which the ES 8 server rejects.

Rolling the workers/scheduler/init-job images back to apache/airflow:3.2.0 (provider 6.5.0) restores task log ingestion immediately against the same ES cluster.

What you think should happen instead?

The Elasticsearch task handler should remain compatible with ES 8.x servers (per the provider docs, ES 8 is still a supported target). Either:

• pin/cap the elasticsearch-py client so the negotiated compatible-with header matches the server major version, or
• expose configuration so operators can force compatible-with=8 (or 7), or
• document that 6.5.1+ requires an ES 9 server and gate the ES 9 client behind a flag.

How to reproduce

1. Run Airflow 3.2.1 (or any Airflow with provider 6.5.1) with the ElasticsearchTaskHandler enabled and write_to_es=True.
2. Point the handler at an Elasticsearch 8.x server.
3. Trigger any task; observe the worker tries to bulk-write logs.

The worker logs the following:

PUT http://<es-host>/_bulk [status:400 duration:0.027s]

Unable to insert logs into Elasticsearch.
Reason: BadRequestError(400, 'media_type_header_exception',
        'Invalid media-type value on headers [Content-Type, Accept]',
        Accept version must be either version 8 or 7, but found 9.
        Accept=application/vnd.elasticsearch+json; compatible-with=9)

  File ".../airflow/providers/elasticsearch/log/es_task_handler.py", line 696, in _write_to_es
      _ = helpers.bulk(self.client, bulk_actions)
  ...
  File ".../elasticsearch/_sync/client/_base.py", line 351, in _perform_request
      raise HTTP_EXCEPTIONS.get(meta.status, ApiError)(...)
elasticsearch.BadRequestError: BadRequestError(400, 'media_type_header_exception', ...)

Operating System

Linux (container image: apache/airflow:3.2.1, python 3.13)

Versions of Apache Airflow Providers

apache-airflow-providers-elasticsearch==6.5.1

Deployment

Official Apache Airflow Helm Chart

Deployment details

• Kubernetes deployment via the official Helm chart
• Elasticsearch server: 8.x (accepts only vnd.elasticsearch+json; compatible-with=7|8)
• Remote logging config:
  • write_to_es=True
  • json_format=True
  • target_index=airflow-logs

Anything else?

• Workaround: pin runtime image to apache/airflow:3.2.0 (provider 6.5.0).
• Likely root cause: PR Support Elasticsearch 9 in Elasticsearch Provider #64070 / provider 6.5.1 bumped elasticsearch>=8.10,<10, and the resolver picks the ES 9 client by default.
• Related upstream behavior: Invalid mediatype with compatible-with throws exception elastic/elasticsearch#79060.

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

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[petercheon]

Filed a draft fix that adds an opt-in [elasticsearch] es_compat_with config option to pin the client's content-negotiation level. With es_compat_with = "8" an elasticsearch>=9 client can keep talking to an Elasticsearch 8.x server without hitting this 400.

PR: #66064 (draft, not auto-detect — see PR body for the rationale)

[petercheon]

Update: my first attempt (#66064) used client.options(headers={...}), which only sets the client's default headers but does not survive elasticsearch-py's per-API-method content-negotiation. Verified with a wire-level capture against elasticsearch-py 9.3.0 that the actual outbound Accept is still compatible-with=9 against an ES 8 server. I've closed that PR.

Replacement: #66065 patches at the transport layer (client.transport.perform_request wrap), covers hooks/elasticsearch.py in addition to logging, and ships wire-level tests that intercept Transport.perform_request and assert on the captured headers.