From c0ef998b717743d36cd1ad19a00ecc18cd440d49 Mon Sep 17 00:00:00 2001
From: Peter Cheon <peter.cheon@kakaocorp.com>
Date: Wed, 29 Apr 2026 11:49:22 +0900
Subject: [PATCH 1/7] fix(elasticsearch): pin compatible-with at the transport
 layer to keep ES 8 servers working
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Since #64070 the provider depends on elasticsearch>=8.10,<10. A default
install resolves to an elasticsearch>=9 Python client, which always
negotiates 'compatible-with=9' on every request. Elasticsearch 8.x
servers reject that with HTTP 400 media_type_header_exception, breaking
remote task log ingestion and both ElasticsearchSQLHook and
ElasticsearchPythonHook against ES 8 clusters.

Add a [elasticsearch] es_compat_with config option that, when set to a
major version string ('7'/'8'/'9'), wraps the client's transport
perform_request so every outbound request carries
'Accept: application/vnd.elasticsearch+json; compatible-with=<major>'
(and the matching '+x-ndjson' form for bulk so streaming bodies still
parse). The wrap is applied at every Elasticsearch client construction
site in the provider:

  - ElasticsearchTaskHandler  (log/es_task_handler.py)
  - ElasticsearchRemoteLogIO  (log/es_task_handler.py)
  - ESConnection              (hooks/elasticsearch.py)
  - ElasticsearchPythonHook   (hooks/elasticsearch.py)

When the option is unset, behavior is unchanged.

Tests assert against what the transport actually sends, not the in-memory
state of the client object. Setting client._headers (which is what
client.options(headers=...) does) is not enough because elasticsearch-py
re-applies its own per-API-method content-negotiation headers right
before the request is sent — only the transport layer sees the final
headers.

Closes: https://github.com/apache/airflow/issues/66063
Supersedes: https://github.com/apache/airflow/pull/66064
---
 providers/elasticsearch/docs/changelog.rst    |  14 ++
 .../elasticsearch/docs/logging/index.rst      |  25 ++++
 providers/elasticsearch/provider.yaml         |  14 ++
 .../providers/elasticsearch/_compat.py        |  82 ++++++++++++
 .../elasticsearch/hooks/elasticsearch.py      |   9 +-
 .../elasticsearch/log/es_task_handler.py      |   5 +-
 .../tests/unit/elasticsearch/test_compat.py   | 122 ++++++++++++++++++
 7 files changed, 266 insertions(+), 5 deletions(-)
 create mode 100644 providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
 create mode 100644 providers/elasticsearch/tests/unit/elasticsearch/test_compat.py

diff --git a/providers/elasticsearch/docs/changelog.rst b/providers/elasticsearch/docs/changelog.rst
index 47f9b3bcbeaa2..5e6a4f695a510 100644
--- a/providers/elasticsearch/docs/changelog.rst
+++ b/providers/elasticsearch/docs/changelog.rst
@@ -35,6 +35,20 @@ a dictionary key in the task-log output when log-hits did not carry a
 ``host`` field. The Elasticsearch client is still connected using the
 full URL, so authentication is unaffected.
 
+A new ``[elasticsearch] es_compat_with`` config option lets operators pin
+the ``compatible-with`` HTTP content-negotiation level used by the
+Elasticsearch client. Since 6.5.1 the provider depends on
+``elasticsearch>=8.10,<10``, and a default install resolves to an
+``elasticsearch>=9`` client which unconditionally negotiates
+``compatible-with=9`` on every request. Elasticsearch 8.x servers reject
+that with HTTP 400 ``media_type_header_exception`` (regression introduced
+by #64070), breaking remote task log ingestion and the SQL/Python hooks
+against ES 8 clusters. Setting ``es_compat_with = "8"`` rewrites the
+client transport so every outbound request carries
+``compatible-with=8`` (and the matching ``+x-ndjson`` form for bulk
+requests), restoring compatibility without dropping ES 9 support. When
+unset, behavior is unchanged.
+
 6.5.3
 .....
 
diff --git a/providers/elasticsearch/docs/logging/index.rst b/providers/elasticsearch/docs/logging/index.rst
index 3254bf5406ccf..499873eac97a5 100644
--- a/providers/elasticsearch/docs/logging/index.rst
+++ b/providers/elasticsearch/docs/logging/index.rst
@@ -93,6 +93,31 @@ Additionally, in the ``elasticsearch_configs`` section, you can pass any paramet
     api_key = "SOMEAPIKEY"
     verify_certs = True
 
+Pinning the ``compatible-with`` content-negotiation level
+'''''''''''''''''''''''''''''''''''''''''''''''''''''''''
+
+Since provider 6.5.1, the Elasticsearch dependency is ``elasticsearch>=8.10,<10``,
+which means a default install resolves to an ``elasticsearch>=9`` Python client.
+That client unconditionally negotiates ``compatible-with=9`` on every request,
+which Elasticsearch 8.x servers reject with HTTP 400
+``media_type_header_exception``. Both the task log writer and the
+``ElasticsearchSQLHook`` / ``ElasticsearchPythonHook`` are affected.
+
+If you need to keep a single Airflow image compatible with an
+``elasticsearch<9`` server, set ``[elasticsearch] es_compat_with`` to the server
+major version. The provider then rewrites the client transport so every outbound
+request carries ``Accept`` / ``Content-Type:
+application/vnd.elasticsearch+json; compatible-with=<major>`` (and the matching
+``+x-ndjson`` form for bulk requests):
+
+.. code-block:: ini
+
+    [elasticsearch]
+    es_compat_with = 8
+
+When the option is unset the client behaves as before (negotiating its own
+major version).
+
 .. _elasticsearch-document-schema:
 
 Expected Elasticsearch document schema
diff --git a/providers/elasticsearch/provider.yaml b/providers/elasticsearch/provider.yaml
index 5fff75d12efec..822580e58c881 100644
--- a/providers/elasticsearch/provider.yaml
+++ b/providers/elasticsearch/provider.yaml
@@ -221,6 +221,20 @@ config:
         type: string
         example: ~
         default: "1000"
+      es_compat_with:
+        description: |
+          Pin the ``compatible-with`` HTTP content-negotiation level used by the
+          Elasticsearch client. Accepts a server major version string (e.g. ``"7"``,
+          ``"8"``, ``"9"``). When unset, the elasticsearch-py client negotiates its
+          own major version, which makes an ``elasticsearch>=9`` client (the default
+          for fresh installs) incompatible with Elasticsearch 8.x servers — every
+          request is rejected with HTTP 400 ``media_type_header_exception``.
+          Setting this option keeps a single Airflow image compatible with both
+          ``elasticsearch<9`` and ``elasticsearch>=9`` servers.
+        version_added: 6.5.4
+        type: string
+        example: "8"
+        default: ""
   elasticsearch_configs:
     description: ~
     options:
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
new file mode 100644
index 0000000000000..f523e72950e26
--- /dev/null
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
@@ -0,0 +1,82 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Helpers shared between the Elasticsearch hooks and log handler.
+
+Currently this exposes a single helper, :func:`apply_compat_with`, that lets the
+provider keep working against an Elasticsearch server whose major version does
+not match the installed ``elasticsearch`` Python client major. See the helper
+docstring for the regression context.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from airflow.providers.common.compat.sdk import conf
+
+if TYPE_CHECKING:
+    import elasticsearch
+
+
+_JSON_MIME = "application/vnd.elasticsearch+json"
+_NDJSON_MIME = "application/vnd.elasticsearch+x-ndjson"
+
+
+def apply_compat_with(client: elasticsearch.Elasticsearch) -> elasticsearch.Elasticsearch:
+    """Pin the ``compatible-with`` HTTP content-negotiation level for ``client``.
+
+    The ``elasticsearch`` Python client always negotiates ``compatible-with=<client_major>``
+    on every request; an Elasticsearch server with a different major version rejects
+    the request with HTTP 400 ``media_type_header_exception``. This is what happens
+    when an ``elasticsearch>=9`` client (the current default) talks to an
+    Elasticsearch 8.x server, which broke remote logging for ES 8 deployments
+    starting with provider 6.5.1.
+
+    When ``[elasticsearch] es_compat_with`` is set to a major version string
+    (e.g. ``"7"``, ``"8"``, ``"9"``) this helper wraps the client's transport
+    so every outbound request carries
+    ``Accept: application/vnd.elasticsearch+json; compatible-with=<major>``
+    and the matching ``Content-Type`` (using the ``+x-ndjson`` form for bulk
+    requests so multi-line bodies still parse on the server).
+
+    When the option is unset the client is returned unchanged and behaves
+    exactly as before.
+    """
+    compat = conf.get("elasticsearch", "es_compat_with", fallback=None)
+    if not compat:
+        return client
+
+    json_media = f"{_JSON_MIME}; compatible-with={compat}"
+    ndjson_media = f"{_NDJSON_MIME}; compatible-with={compat}"
+
+    transport = client.transport
+    original_perform_request = transport.perform_request
+
+    def perform_request(method, target, *, body=None, headers=None, **kwargs):  # type: ignore[no-untyped-def]
+        merged = dict(headers) if headers else {}
+        if merged.get("accept"):
+            merged["accept"] = json_media
+        content_type = merged.get("content-type") or ""
+        if "ndjson" in content_type or "x-ndjson" in content_type:
+            merged["content-type"] = ndjson_media
+        elif content_type:
+            merged["content-type"] = json_media
+        return original_perform_request(method, target, body=body, headers=merged, **kwargs)
+
+    transport.perform_request = perform_request
+    return client
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
index e8f0cfad00ce6..5e04f327c7b71 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
@@ -27,6 +27,7 @@
 
 from airflow.providers.common.compat.sdk import BaseHook
 from airflow.providers.common.sql.hooks.sql import DbApiHook
+from airflow.providers.elasticsearch._compat import apply_compat_with
 
 if TYPE_CHECKING:
     from elastic_transport import ObjectApiResponse
@@ -135,9 +136,11 @@ def __init__(
         netloc = f"{host}:{port}"
         self.url = parse.urlunparse((scheme, netloc, "/", None, None, None))
         if user and password:
-            self.es = Elasticsearch(self.url, basic_auth=(user, password), **kwargs)
+            self.es = apply_compat_with(
+                Elasticsearch(self.url, basic_auth=(user, password), **kwargs)
+            )
         else:
-            self.es = Elasticsearch(self.url, **kwargs)
+            self.es = apply_compat_with(Elasticsearch(self.url, **kwargs))
 
     def cursor(self) -> ElasticsearchSQLCursor:
         return ElasticsearchSQLCursor(self.es, **self.kwargs)
@@ -247,7 +250,7 @@ def __init__(self, hosts: list[Any], es_conn_args: dict | None = None):
 
     def _get_elastic_connection(self):
         """Return the Elasticsearch client."""
-        client = Elasticsearch(self.hosts, **self.es_conn_args)
+        client = apply_compat_with(Elasticsearch(self.hosts, **self.es_conn_args))
 
         return client
 
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
index a8c676a8699c2..fc0b7cbea950d 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/log/es_task_handler.py
@@ -43,6 +43,7 @@
 import airflow.logging_config as alc
 from airflow.models.dagrun import DagRun
 from airflow.providers.common.compat.sdk import conf
+from airflow.providers.elasticsearch._compat import apply_compat_with
 from airflow.providers.elasticsearch.log.es_json_formatter import ElasticsearchJSONFormatter
 from airflow.providers.elasticsearch.log.es_response import ElasticSearchResponse, Hit, resolve_nested
 from airflow.providers.elasticsearch.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS
@@ -269,7 +270,7 @@ def __init__(
         )
         self.closed = False
 
-        self.client = elasticsearch.Elasticsearch(self.host, **es_kwargs)
+        self.client = apply_compat_with(elasticsearch.Elasticsearch(self.host, **es_kwargs))
         # in airflow.cfg, host of elasticsearch has to be http://dockerhostXxxx:9200
 
         self.frontend = frontend
@@ -653,7 +654,7 @@ class ElasticsearchRemoteLogIO(LoggingMixin):  # noqa: D101
 
     def __attrs_post_init__(self):
         es_kwargs = get_es_kwargs_from_config()
-        self.client = elasticsearch.Elasticsearch(self.host, **es_kwargs)
+        self.client = apply_compat_with(elasticsearch.Elasticsearch(self.host, **es_kwargs))
         self.index_patterns_callable = conf.get("elasticsearch", "index_patterns_callable", fallback="")
         self.PAGE = 0
         self.MAX_LINE_PER_PAGE = conf.getint("elasticsearch", "max_lines_per_page", fallback=1000)
diff --git a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py b/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
new file mode 100644
index 0000000000000..35ac1e2fc1bf5
--- /dev/null
+++ b/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
@@ -0,0 +1,122 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""Wire-level tests for ``airflow.providers.elasticsearch._compat.apply_compat_with``.
+
+These tests intercept ``elastic_transport.Transport.perform_request`` so they
+observe the exact ``Accept`` / ``Content-Type`` headers the helper produces.
+Asserting on ``client._headers`` (or any other in-memory state on the client)
+would not be enough: the elasticsearch-py client always re-applies its own
+per-API-method content-negotiation headers right before the request goes out,
+which is the very behaviour this helper has to override. Only what the
+``Transport`` sees on the wire matters.
+"""
+
+from __future__ import annotations
+
+import contextlib
+
+import pytest
+from elastic_transport import Transport
+from elasticsearch import Elasticsearch
+
+from airflow.providers.elasticsearch._compat import apply_compat_with
+
+from tests_common.test_utils.config import conf_vars
+
+
+def _trigger_calls(client: Elasticsearch) -> None:
+    """Drive ``search``, ``info`` and ``bulk`` against the spy transport.
+
+    Each call hits the spy and raises; we swallow that and rely on the spy's
+    ``captured`` list to make assertions.
+    """
+    for action in (
+        lambda: client.search(index="airflow-logs", query={"match_all": {}}),
+        lambda: client.info(),
+        lambda: client.bulk(operations=[{"index": {"_index": "x"}}, {"hello": "world"}]),
+    ):
+        with contextlib.suppress(RuntimeError):
+            action()
+
+
+@pytest.fixture
+def wire_capture(monkeypatch):
+    """Replace ``Transport.perform_request`` with a recording spy.
+
+    The spy is installed at the *class* level (not on an instance) so it is
+    picked up by both the original transport and by the wrapper produced by
+    :func:`apply_compat_with`. The wrapper resolves the original
+    ``perform_request`` at wrap time, so the spy must already be in place when
+    ``apply_compat_with`` runs.
+    """
+    captured: list[dict] = []
+
+    def spy(self, method, target, *, body=None, headers=None, **kwargs):
+        captured.append(
+            {"method": method, "target": target, "headers": dict(headers or {})}
+        )
+        raise RuntimeError("captured")
+
+    monkeypatch.setattr(Transport, "perform_request", spy)
+    return captured
+
+
+@pytest.mark.parametrize("unset_value", ["", None])
+def test_apply_compat_with_unset_does_not_wrap_transport(unset_value):
+    """When the option is unset the helper returns the client untouched."""
+    client = Elasticsearch("http://localhost:9200")
+    original = client.transport.perform_request
+    with conf_vars({("elasticsearch", "es_compat_with"): unset_value}):
+        same = apply_compat_with(client)
+    assert same is client
+    assert client.transport.perform_request is original
+
+
+def test_apply_compat_with_pins_compatible_with_8(wire_capture):
+    """With ``es_compat_with = "8"`` every outbound call carries ``compatible-with=8``."""
+    with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    _trigger_calls(client)
+
+    assert {c["method"] for c in wire_capture} == {"POST", "GET", "PUT"}
+    expected_json = "application/vnd.elasticsearch+json; compatible-with=8"
+    expected_ndjson = "application/vnd.elasticsearch+x-ndjson; compatible-with=8"
+
+    by_method = {c["method"]: c for c in wire_capture}
+    assert by_method["POST"]["headers"]["accept"] == expected_json
+    assert by_method["POST"]["headers"]["content-type"] == expected_json
+    assert by_method["GET"]["headers"]["accept"] == expected_json
+    # ``info()`` does not send a body, so content-type is absent on the wire
+    assert by_method["GET"]["headers"].get("content-type") in (None, "")
+    assert by_method["PUT"]["headers"]["accept"] == expected_json
+    # bulk preserves the ndjson form so the server can stream the body
+    assert by_method["PUT"]["headers"]["content-type"] == expected_ndjson
+
+
+def test_apply_compat_with_pins_compatible_with_7(wire_capture):
+    """The helper accepts arbitrary major version strings, not just ``"8"``."""
+    with conf_vars({("elasticsearch", "es_compat_with"): "7"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    _trigger_calls(client)
+    assert wire_capture, "spy should have captured at least one call"
+    assert all(
+        c["headers"]["accept"] == "application/vnd.elasticsearch+json; compatible-with=7"
+        for c in wire_capture
+    )

From bf7cbc1784c495ac9b3b584625897f577949b0be Mon Sep 17 00:00:00 2001
From: Peter Cheon <peter.cheon@kakaocorp.com>
Date: Wed, 29 Apr 2026 13:10:17 +0900
Subject: [PATCH 2/7] =?UTF-8?q?fixup(elasticsearch):=20tighten=20apply=5Fc?=
 =?UTF-8?q?ompat=5Fwith=20=E2=80=94=20idempotent=20guard,=20simpler=20ct?=
 =?UTF-8?q?=20check,=20correct=20test=20assertion?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- idempotent guard: skip second wrap by checking transport.__dict__ for an
  existing instance attribute. Repeated apply_compat_with calls (e.g.
  hook reuse paths) are now true no-ops.
- content-type check simplified: '"ndjson" in ct' already matches both
  'ndjson' and '+x-ndjson' so the redundant 'x-ndjson' branch is dropped.
- unset-case test was using 'transport.perform_request is original' which
  would fail even when nothing was wrapped, because attribute access on a
  bound method produces a fresh wrapper object every time. Switched to
  inspecting transport.__dict__ for the 'perform_request' key, which
  precisely tracks whether the helper installed an instance override.
- new test_apply_compat_with_is_idempotent asserts the guard above.
---
 .../providers/elasticsearch/_compat.py        | 12 +++++++--
 .../tests/unit/elasticsearch/test_compat.py   | 25 ++++++++++++++++---
 2 files changed, 32 insertions(+), 5 deletions(-)

diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
index f523e72950e26..8d03d8facf19c 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
@@ -61,10 +61,15 @@ def apply_compat_with(client: elasticsearch.Elasticsearch) -> elasticsearch.Elas
     if not compat:
         return client
 
+    transport = client.transport
+    if "perform_request" in transport.__dict__:
+        # Already wrapped on this transport instance — no-op so repeated calls
+        # to ``apply_compat_with`` (e.g. across hook reuse) stay idempotent.
+        return client
+
     json_media = f"{_JSON_MIME}; compatible-with={compat}"
     ndjson_media = f"{_NDJSON_MIME}; compatible-with={compat}"
 
-    transport = client.transport
     original_perform_request = transport.perform_request
 
     def perform_request(method, target, *, body=None, headers=None, **kwargs):  # type: ignore[no-untyped-def]
@@ -72,7 +77,10 @@ def perform_request(method, target, *, body=None, headers=None, **kwargs):  # ty
         if merged.get("accept"):
             merged["accept"] = json_media
         content_type = merged.get("content-type") or ""
-        if "ndjson" in content_type or "x-ndjson" in content_type:
+        # ``+x-ndjson`` is the only streaming form elasticsearch-py uses today
+        # (bulk requests). ``"ndjson" in ct`` already matches both ``ndjson``
+        # and ``x-ndjson``, so a single check is enough.
+        if "ndjson" in content_type:
             merged["content-type"] = ndjson_media
         elif content_type:
             merged["content-type"] = json_media
diff --git a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py b/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
index 35ac1e2fc1bf5..479851eea0035 100644
--- a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
+++ b/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
@@ -78,13 +78,22 @@ def spy(self, method, target, *, body=None, headers=None, **kwargs):
 
 @pytest.mark.parametrize("unset_value", ["", None])
 def test_apply_compat_with_unset_does_not_wrap_transport(unset_value):
-    """When the option is unset the helper returns the client untouched."""
+    """When the option is unset the helper returns the client untouched.
+
+    The wrap installs ``perform_request`` as an instance attribute on the
+    transport, so the most precise way to assert the helper is a no-op is to
+    check that no such instance attribute was set (i.e. lookup still resolves
+    to the class method). Identity (``is``) comparison on
+    ``transport.perform_request`` would not work even in the no-op case
+    because attribute access on a bound method produces a fresh wrapper each
+    time.
+    """
     client = Elasticsearch("http://localhost:9200")
-    original = client.transport.perform_request
+    assert "perform_request" not in client.transport.__dict__
     with conf_vars({("elasticsearch", "es_compat_with"): unset_value}):
         same = apply_compat_with(client)
     assert same is client
-    assert client.transport.perform_request is original
+    assert "perform_request" not in client.transport.__dict__
 
 
 def test_apply_compat_with_pins_compatible_with_8(wire_capture):
@@ -120,3 +129,13 @@ def test_apply_compat_with_pins_compatible_with_7(wire_capture):
         c["headers"]["accept"] == "application/vnd.elasticsearch+json; compatible-with=7"
         for c in wire_capture
     )
+
+
+def test_apply_compat_with_is_idempotent():
+    """Calling ``apply_compat_with`` twice on the same client only wraps once."""
+    with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+        first_wrapper = client.transport.__dict__["perform_request"]
+        apply_compat_with(client)
+        second_wrapper = client.transport.__dict__["perform_request"]
+    assert first_wrapper is second_wrapper

From 780f963dd42dc5d179541ef74fbef5a5b79b2f0e Mon Sep 17 00:00:00 2001
From: Peter <Peter.cheon@kakaocorp.com>
Date: Wed, 29 Apr 2026 14:59:24 +0900
Subject: [PATCH 3/7] fixup(elasticsearch): mirror upstream selective mimetype
 rewrite

The previous wrapper unconditionally overwrote the entire `Accept` header
to `application/vnd.elasticsearch+json; compatible-with=<major>` whenever
one was present. That is too aggressive: elasticsearch-py emits
non-JSON `Accept` values for several APIs that still need to flow through
the same transport. Notably:

- `client.cat.help()` sends `Accept: text/plain`.
- All other `client.cat.*` endpoints send `Accept: text/plain,application/json`.
- Search-MVT endpoints send `Accept: application/vnd.mapbox-vector-tile`.

After the previous wrap every one of those calls went on the wire as
plain `application/vnd.elasticsearch+json; compatible-with=<N>`, silently
turning cat responses into JSON for any operator using
`ElasticsearchPythonHook.get_conn()` to call cat APIs.

Mirror upstream's own `mimetype_header_to_compat` instead: only
`application/(json|x-ndjson|vnd.mapbox-vector-tile)` parts of the header
get the `compatible-with=<configured>` suffix, anything else is left
verbatim. The regex also matches the already-rewritten
`application/vnd.elasticsearch+<x>; compatible-with=<N>` form that
elasticsearch-py 9.x ships before the transport sees the request, so the
configured major actually replaces the client default major on the wire
(verified with a Transport spy against elasticsearch-py 9.3.0).

Two adjacent hardenings while we are in here:

- Strip whitespace from the config value and reject anything that is not a
  positive integer string with `AirflowConfigException` at construction
  time, so a typo like `es_compat_with = 'v8'` fails fast in the worker
  startup log instead of returning a 400 storm per request.
- Walk header keys case-insensitively, so a future `elastic_transport`
  that forwards PascalCase `Accept` / `Content-Type` keys cannot silently
  bypass the rewrite.

Tests: add wire-level cases for cat APIs (`text/plain` preserved,
`text/plain,application/...` partial rewrite), PascalCase headers,
whitespace stripping, non-numeric major rejection, and a direct
`conf.get -> None` branch (the existing parametrize folds into the
provider yaml default `""` via `conf_vars`).
---
 .../providers/elasticsearch/_compat.py        |  58 ++++++----
 .../tests/unit/elasticsearch/test_compat.py   | 103 ++++++++++++++++++
 2 files changed, 140 insertions(+), 21 deletions(-)

diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
index 8d03d8facf19c..1343dc1faff40 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
@@ -25,16 +25,30 @@
 
 from __future__ import annotations
 
+import re
 from typing import TYPE_CHECKING
 
+from airflow.exceptions import AirflowConfigException
 from airflow.providers.common.compat.sdk import conf
 
 if TYPE_CHECKING:
     import elasticsearch
 
 
-_JSON_MIME = "application/vnd.elasticsearch+json"
-_NDJSON_MIME = "application/vnd.elasticsearch+x-ndjson"
+# Matches the JSON / NDJSON / mapbox vector-tile mimetypes the ``elasticsearch``
+# client negotiates, in either form: the raw ``application/json`` (what the
+# generated client code writes into ``__headers``) and the already-rewritten
+# ``application/vnd.elasticsearch+json; compatible-with=<N>`` (what
+# ``mimetype_header_to_compat`` in ``elasticsearch/_sync/client/_base.py``
+# emits before handing the request to the transport). Both forms must be
+# rewritten to ``compatible-with=<configured>``; anything else (notably
+# ``text/plain`` used by the cat APIs) is left intact, mirroring upstream's
+# selective substitution behaviour.
+_COMPAT_MIMETYPE_RE = re.compile(
+    r"application/(?:vnd\.elasticsearch\+)?(json|x-ndjson|vnd\.mapbox-vector-tile)"
+    r"(?:\s*;\s*compatible-with=\d+)?"
+)
+_COMPAT_MAJOR_RE = re.compile(r"^\d+$")
 
 
 def apply_compat_with(client: elasticsearch.Elasticsearch) -> elasticsearch.Elasticsearch:
@@ -49,17 +63,24 @@ def apply_compat_with(client: elasticsearch.Elasticsearch) -> elasticsearch.Elas
 
     When ``[elasticsearch] es_compat_with`` is set to a major version string
     (e.g. ``"7"``, ``"8"``, ``"9"``) this helper wraps the client's transport
-    so every outbound request carries
-    ``Accept: application/vnd.elasticsearch+json; compatible-with=<major>``
-    and the matching ``Content-Type`` (using the ``+x-ndjson`` form for bulk
-    requests so multi-line bodies still parse on the server).
+    so every outbound request rewrites the ``compatible-with=<N>`` parameter on
+    the JSON / NDJSON / mapbox vector-tile parts of the ``Accept`` and
+    ``Content-Type`` headers. Non-JSON parts (notably ``text/plain`` used by
+    the cat APIs) are preserved verbatim, mirroring how elasticsearch-py's own
+    ``mimetype_header_to_compat`` handles content negotiation.
 
     When the option is unset the client is returned unchanged and behaves
     exactly as before.
     """
-    compat = conf.get("elasticsearch", "es_compat_with", fallback=None)
+    raw = conf.get("elasticsearch", "es_compat_with", fallback=None)
+    compat = (raw or "").strip()
     if not compat:
         return client
+    if not _COMPAT_MAJOR_RE.match(compat):
+        raise AirflowConfigException(
+            "[elasticsearch] es_compat_with must be a positive integer major version "
+            f"(e.g. '7', '8', '9'); got {raw!r}."
+        )
 
     transport = client.transport
     if "perform_request" in transport.__dict__:
@@ -67,23 +88,18 @@ def apply_compat_with(client: elasticsearch.Elasticsearch) -> elasticsearch.Elas
         # to ``apply_compat_with`` (e.g. across hook reuse) stay idempotent.
         return client
 
-    json_media = f"{_JSON_MIME}; compatible-with={compat}"
-    ndjson_media = f"{_NDJSON_MIME}; compatible-with={compat}"
-
+    sub = rf"application/vnd.elasticsearch+\g<1>; compatible-with={compat}"
     original_perform_request = transport.perform_request
 
     def perform_request(method, target, *, body=None, headers=None, **kwargs):  # type: ignore[no-untyped-def]
-        merged = dict(headers) if headers else {}
-        if merged.get("accept"):
-            merged["accept"] = json_media
-        content_type = merged.get("content-type") or ""
-        # ``+x-ndjson`` is the only streaming form elasticsearch-py uses today
-        # (bulk requests). ``"ndjson" in ct`` already matches both ``ndjson``
-        # and ``x-ndjson``, so a single check is enough.
-        if "ndjson" in content_type:
-            merged["content-type"] = ndjson_media
-        elif content_type:
-            merged["content-type"] = json_media
+        if not headers:
+            return original_perform_request(method, target, body=body, headers=headers, **kwargs)
+        # Walk every key case-insensitively so a future elastic_transport that
+        # forwards PascalCase headers does not silently bypass the rewrite.
+        merged = dict(headers)
+        for key in list(merged):
+            if key.lower() in ("accept", "content-type") and merged[key]:
+                merged[key] = _COMPAT_MIMETYPE_RE.sub(sub, merged[key])
         return original_perform_request(method, target, body=body, headers=merged, **kwargs)
 
     transport.perform_request = perform_request
diff --git a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py b/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
index 479851eea0035..0acf59d139443 100644
--- a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
+++ b/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
@@ -34,6 +34,7 @@
 from elastic_transport import Transport
 from elasticsearch import Elasticsearch
 
+from airflow.exceptions import AirflowConfigException
 from airflow.providers.elasticsearch._compat import apply_compat_with
 
 from tests_common.test_utils.config import conf_vars
@@ -87,6 +88,12 @@ def test_apply_compat_with_unset_does_not_wrap_transport(unset_value):
     ``transport.perform_request`` would not work even in the no-op case
     because attribute access on a bound method produces a fresh wrapper each
     time.
+
+    Note: ``conf_vars`` removes the override when the value is ``None``, so
+    both parametrized cases ultimately resolve to the provider yaml default
+    (``""``). The parametrize is kept to document that callers passing either
+    sentinel get the no-op path; the actual ``None`` branch in the helper is
+    covered by ``test_apply_compat_with_unset_via_missing_conf`` below.
     """
     client = Elasticsearch("http://localhost:9200")
     assert "perform_request" not in client.transport.__dict__
@@ -96,6 +103,16 @@ def test_apply_compat_with_unset_does_not_wrap_transport(unset_value):
     assert "perform_request" not in client.transport.__dict__
 
 
+def test_apply_compat_with_unset_via_missing_conf(monkeypatch):
+    """Cover the ``conf.get`` returning ``None`` branch directly."""
+    from airflow.providers.elasticsearch import _compat
+
+    monkeypatch.setattr(_compat.conf, "get", lambda *args, **kwargs: None)
+    client = Elasticsearch("http://localhost:9200")
+    assert apply_compat_with(client) is client
+    assert "perform_request" not in client.transport.__dict__
+
+
 def test_apply_compat_with_pins_compatible_with_8(wire_capture):
     """With ``es_compat_with = "8"`` every outbound call carries ``compatible-with=8``."""
     with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
@@ -131,6 +148,92 @@ def test_apply_compat_with_pins_compatible_with_7(wire_capture):
     )
 
 
+def test_apply_compat_with_preserves_text_plain_for_cat_apis(wire_capture):
+    """Cat APIs send ``Accept: text/plain[,application/json]``; the wrapper must
+    preserve the ``text/plain`` part. Earlier revisions of the helper unconditionally
+    rewrote ``accept`` to ``application/vnd.elasticsearch+json; compatible-with=N``,
+    which silently turned every ``cat.*`` response into JSON instead of plain text.
+
+    We mirror elasticsearch-py's own ``mimetype_header_to_compat`` (only
+    ``application/(json|x-ndjson|vnd.mapbox-vector-tile)`` parts get the
+    ``compatible-with`` suffix), so this test fails fast if anyone reverts to the
+    blanket overwrite.
+    """
+    with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    for action in (lambda: client.cat.help(), lambda: client.cat.indices()):
+        with contextlib.suppress(RuntimeError):
+            action()
+
+    accepts = [c["headers"].get("accept") for c in wire_capture]
+    # ``cat.help`` ships ``text/plain`` only; it must come through verbatim.
+    assert "text/plain" in accepts, accepts
+    # ``cat.indices`` ships ``text/plain,application/json``; the JSON half gets
+    # the ``compatible-with=8`` suffix, the text/plain half stays put.
+    assert any(
+        a and a.startswith("text/plain,application/vnd.elasticsearch+json; compatible-with=8")
+        for a in accepts
+    ), accepts
+
+
+def test_apply_compat_with_handles_pascal_case_headers(monkeypatch):
+    """Defensive: if ``elastic_transport`` ever forwards PascalCase header keys,
+    the rewrite must still apply (a lowercase-only ``dict.get`` would silently
+    no-op and let ``compatible-with=<client_major>`` ship to the server).
+    """
+    seen: dict = {}
+
+    def spy(self, method, target, *, body=None, headers=None, **kwargs):
+        seen["headers"] = dict(headers or {})
+        raise RuntimeError("captured")
+
+    monkeypatch.setattr(Transport, "perform_request", spy)
+
+    with conf_vars({("elasticsearch", "es_compat_with"): "8"}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    # Drive the wrapper with PascalCase keys directly — bypassing the
+    # ``_BaseClient.perform_request`` normalization.
+    with contextlib.suppress(RuntimeError):
+        client.transport.perform_request(
+            "GET",
+            "/",
+            headers={"Accept": "application/json", "Content-Type": "application/json"},
+        )
+
+    assert seen["headers"]["Accept"] == "application/vnd.elasticsearch+json; compatible-with=8"
+    assert seen["headers"]["Content-Type"] == "application/vnd.elasticsearch+json; compatible-with=8"
+
+
+def test_apply_compat_with_strips_whitespace_in_config(wire_capture):
+    """Operators occasionally write ``es_compat_with = " 8"``; the helper must
+    strip whitespace before interpolating into the wire header, otherwise the
+    server returns 400 and the helper fails open in a confusing way.
+    """
+    with conf_vars({("elasticsearch", "es_compat_with"): " 8 "}):
+        client = apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+    with contextlib.suppress(RuntimeError):
+        client.search(index="airflow-logs", query={"match_all": {}})
+
+    assert wire_capture[-1]["headers"]["accept"] == (
+        "application/vnd.elasticsearch+json; compatible-with=8"
+    )
+
+
+@pytest.mark.parametrize("bad_value", ["v8", "8.0", "abc", "8;9"])
+def test_apply_compat_with_rejects_non_numeric_major(bad_value):
+    """A non-numeric ``es_compat_with`` would otherwise produce malformed wire
+    headers (``compatible-with=v8``) and a per-request 400 storm. Fail fast at
+    construction time with a config exception so the misconfiguration is
+    obvious in the worker startup log.
+    """
+    with conf_vars({("elasticsearch", "es_compat_with"): bad_value}):
+        with pytest.raises(AirflowConfigException, match="es_compat_with"):
+            apply_compat_with(Elasticsearch("http://localhost:9200"))
+
+
 def test_apply_compat_with_is_idempotent():
     """Calling ``apply_compat_with`` twice on the same client only wraps once."""
     with conf_vars({("elasticsearch", "es_compat_with"): "8"}):

From 879b864e94f2bb4a6659c28f5594d6b6c67fe483 Mon Sep 17 00:00:00 2001
From: Peter <Peter.cheon@kakaocorp.com>
Date: Wed, 29 Apr 2026 15:11:25 +0900
Subject: [PATCH 4/7] fixup(elasticsearch): import AirflowConfigException via
 common.compat.sdk

The rest of this module already routes airflow imports through the
`common.compat.sdk` shim (`conf` lives there), and the shim explicitly
exports `AirflowConfigException` so the same provider build can target
both Airflow 2 (`airflow.exceptions`) and Airflow 3 (`airflow.sdk.exceptions`).
Switch the new exception import to the same shim so we don't pin to
`airflow.exceptions` and silently break the Airflow 3 import path.
---
 .../src/airflow/providers/elasticsearch/_compat.py             | 3 +--
 .../elasticsearch/tests/unit/elasticsearch/test_compat.py      | 2 +-
 2 files changed, 2 insertions(+), 3 deletions(-)

diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
index 1343dc1faff40..cec4c151cdcae 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
@@ -28,8 +28,7 @@
 import re
 from typing import TYPE_CHECKING
 
-from airflow.exceptions import AirflowConfigException
-from airflow.providers.common.compat.sdk import conf
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
 
 if TYPE_CHECKING:
     import elasticsearch
diff --git a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py b/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
index 0acf59d139443..5c8a717ac582a 100644
--- a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
+++ b/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
@@ -34,7 +34,7 @@
 from elastic_transport import Transport
 from elasticsearch import Elasticsearch
 
-from airflow.exceptions import AirflowConfigException
+from airflow.providers.common.compat.sdk import AirflowConfigException
 from airflow.providers.elasticsearch._compat import apply_compat_with
 
 from tests_common.test_utils.config import conf_vars

From 028717f85af75eefa45d07877ace306896e49a1f Mon Sep 17 00:00:00 2001
From: Peter <Peter.cheon@kakaocorp.com>
Date: Mon, 11 May 2026 08:03:24 +0900
Subject: [PATCH 5/7] fixup(elasticsearch): satisfy CI lint, mypy and
 project-structure tests
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

CI on the latest `main` merge surfaced four failures, all mechanical and
fixed in this commit:

1. **Static checks** (ruff / ruff-format / autogen):
   - `_compat.py` — D205 docstring rule wants the summary line on its own
     line, both for the module docstring and for `apply_compat_with`'s
     docstring. Reformatted both.
   - `hooks/elasticsearch.py` — collapsed the multi-line
     `apply_compat_with(Elasticsearch(...))` call into a single line
     (now under the line-length cap thanks to the basic_auth tuple sitting
     inside the existing parens).
   - `tests/.../test__compat.py` — collapsed two over-wrapped expressions
     (`captured.append({...})` in the spy, and the
     `assert wire_capture[-1][...] == "..."` in
     `test_apply_compat_with_strips_whitespace_in_config`).
   - `get_provider_info.py` — the autogenerated mirror of `provider.yaml`
     was missing the new `es_compat_with` config option entry. Added it
     with the same description / version_added / type / example / default
     as the yaml.

2. **MyPy providers** (`Cannot assign to a method [method-assign]`):
   - `transport.perform_request = perform_request` (instance-level
     assignment) is rejected by mypy because elastic_transport's
     `Transport.perform_request` is bound at the class. Switched to
     `setattr(transport, "perform_request", perform_request)`, which
     mypy accepts and which preserves the exact same runtime behaviour
     (the idempotency guard at the top of the function still inspects
     `transport.__dict__["perform_request"]`, so repeat calls remain
     no-ops).

3. **Non-DB tests core** and **Low dep tests core**
   (`test_project_structure.py::test_providers_modules_should_have_tests`):
   - The structure check expects the test file for source `_compat.py`
     (note the leading underscore) to be named `test__compat.py` (two
     underscores: `test_` + `_compat`). Renamed the file from
     `test_compat.py` → `test__compat.py` via `git mv` so the rest of
     git history follows.

Re-validated locally:
- `ruff check` and `ruff format --check` pass on all four files.
- mypy on `_compat.py` no longer reports the `method-assign` error
  (only an unrelated `airflow.__version__` attr-defined error from
  running mypy outside a real Airflow install — Airflow CI runs against
  an installed Airflow so this does not surface there).
- Wire-level regression matrix re-run with elasticsearch-py 9.3.0 and
  the `setattr` variant: cat.help `text/plain` preserved, cat.indices
  partial rewrite preserved, search/bulk Accept and Content-Type
  rewritten to compat=8, idempotency guard still triggers, bad values
  rejected. 7/7 PASS.
---
 .../src/airflow/providers/elasticsearch/_compat.py   | 12 +++++++++---
 .../providers/elasticsearch/get_provider_info.py     |  7 +++++++
 .../providers/elasticsearch/hooks/elasticsearch.py   |  4 +---
 .../{test_compat.py => test__compat.py}              |  8 ++------
 4 files changed, 19 insertions(+), 12 deletions(-)
 rename providers/elasticsearch/tests/unit/elasticsearch/{test_compat.py => test__compat.py} (97%)

diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
index cec4c151cdcae..b55cd24823305 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
@@ -15,7 +15,8 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-"""Helpers shared between the Elasticsearch hooks and log handler.
+"""
+Helpers shared between the Elasticsearch hooks and log handler.
 
 Currently this exposes a single helper, :func:`apply_compat_with`, that lets the
 provider keep working against an Elasticsearch server whose major version does
@@ -51,7 +52,8 @@
 
 
 def apply_compat_with(client: elasticsearch.Elasticsearch) -> elasticsearch.Elasticsearch:
-    """Pin the ``compatible-with`` HTTP content-negotiation level for ``client``.
+    """
+    Pin the ``compatible-with`` HTTP content-negotiation level for ``client``.
 
     The ``elasticsearch`` Python client always negotiates ``compatible-with=<client_major>``
     on every request; an Elasticsearch server with a different major version rejects
@@ -101,5 +103,9 @@ def perform_request(method, target, *, body=None, headers=None, **kwargs):  # ty
                 merged[key] = _COMPAT_MIMETYPE_RE.sub(sub, merged[key])
         return original_perform_request(method, target, body=body, headers=merged, **kwargs)
 
-    transport.perform_request = perform_request
+    # ``setattr`` instead of direct attribute assignment so mypy does not flag a
+    # ``method-assign`` error — we are *intentionally* shadowing the bound method
+    # at the instance level (the idempotency guard above checks the instance
+    # ``__dict__``).
+    setattr(transport, "perform_request", perform_request)
     return client
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py
index 2d357cecb1944..b0853a98580cd 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/get_provider_info.py
@@ -151,6 +151,13 @@ def get_provider_info():
                         "example": None,
                         "default": "1000",
                     },
+                    "es_compat_with": {
+                        "description": 'Pin the ``compatible-with`` HTTP content-negotiation level used by the\nElasticsearch client. Accepts a server major version string (e.g. ``"7"``,\n``"8"``, ``"9"``). When unset, the elasticsearch-py client negotiates its\nown major version, which makes an ``elasticsearch>=9`` client (the default\nfor fresh installs) incompatible with Elasticsearch 8.x servers — every\nrequest is rejected with HTTP 400 ``media_type_header_exception``.\nSetting this option keeps a single Airflow image compatible with both\n``elasticsearch<9`` and ``elasticsearch>=9`` servers.\n',
+                        "version_added": "6.5.4",
+                        "type": "string",
+                        "example": "8",
+                        "default": "",
+                    },
                 },
             },
             "elasticsearch_configs": {
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
index 5e04f327c7b71..e8d1610d64709 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
@@ -136,9 +136,7 @@ def __init__(
         netloc = f"{host}:{port}"
         self.url = parse.urlunparse((scheme, netloc, "/", None, None, None))
         if user and password:
-            self.es = apply_compat_with(
-                Elasticsearch(self.url, basic_auth=(user, password), **kwargs)
-            )
+            self.es = apply_compat_with(Elasticsearch(self.url, basic_auth=(user, password), **kwargs))
         else:
             self.es = apply_compat_with(Elasticsearch(self.url, **kwargs))
 
diff --git a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py b/providers/elasticsearch/tests/unit/elasticsearch/test__compat.py
similarity index 97%
rename from providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
rename to providers/elasticsearch/tests/unit/elasticsearch/test__compat.py
index 5c8a717ac582a..7f81f9b92f099 100644
--- a/providers/elasticsearch/tests/unit/elasticsearch/test_compat.py
+++ b/providers/elasticsearch/tests/unit/elasticsearch/test__compat.py
@@ -68,9 +68,7 @@ def wire_capture(monkeypatch):
     captured: list[dict] = []
 
     def spy(self, method, target, *, body=None, headers=None, **kwargs):
-        captured.append(
-            {"method": method, "target": target, "headers": dict(headers or {})}
-        )
+        captured.append({"method": method, "target": target, "headers": dict(headers or {})})
         raise RuntimeError("captured")
 
     monkeypatch.setattr(Transport, "perform_request", spy)
@@ -217,9 +215,7 @@ def test_apply_compat_with_strips_whitespace_in_config(wire_capture):
     with contextlib.suppress(RuntimeError):
         client.search(index="airflow-logs", query={"match_all": {}})
 
-    assert wire_capture[-1]["headers"]["accept"] == (
-        "application/vnd.elasticsearch+json; compatible-with=8"
-    )
+    assert wire_capture[-1]["headers"]["accept"] == "application/vnd.elasticsearch+json; compatible-with=8"
 
 
 @pytest.mark.parametrize("bad_value", ["v8", "8.0", "abc", "8;9"])

From 5d8b8f907163b7ff39f797e77088375837a9c691 Mon Sep 17 00:00:00 2001
From: Peter <Peter.cheon@kakaocorp.com>
Date: Tue, 12 May 2026 09:09:22 +0900
Subject: [PATCH 6/7] Address review feedback: tolerant transport wrapper +
 clarified docs

- Refactor apply_compat_with to use functools.wraps + *args, **kwargs so
  the wrapper survives future elastic_transport perform_request signature
  changes (new keyword-only params, reordered positionals) and preserves
  __name__/__doc__/__wrapped__ for introspection.

- Extend the es_compat_with docs entry with explicit valid-value rules
  and a note that the fix is installed at the transport layer and
  therefore overrides elasticsearch-py's per-API-method header
  negotiation (constructor headers= does not work for this purpose).
---
 providers/elasticsearch/docs/logging/index.rst  | 17 +++++++++++++++++
 .../airflow/providers/elasticsearch/_compat.py  | 14 +++++++++++---
 2 files changed, 28 insertions(+), 3 deletions(-)

diff --git a/providers/elasticsearch/docs/logging/index.rst b/providers/elasticsearch/docs/logging/index.rst
index 499873eac97a5..5cfae8d6230da 100644
--- a/providers/elasticsearch/docs/logging/index.rst
+++ b/providers/elasticsearch/docs/logging/index.rst
@@ -115,6 +115,23 @@ application/vnd.elasticsearch+json; compatible-with=<major>`` (and the matching
     [elasticsearch]
     es_compat_with = 8
 
+Only a positive integer major version is accepted (``"7"``, ``"8"``, ``"9"``);
+any other value (e.g. ``"v8"``, ``"8.0"``) fails fast with an
+``AirflowConfigException`` at client construction time so the misconfiguration
+is obvious in the worker startup log instead of producing a per-request 400
+storm.
+
+.. note::
+
+   The fix is installed at the **transport layer** (a wrapper around
+   ``client.transport.perform_request``) and therefore overrides the
+   per-API-method ``Accept`` / ``Content-Type`` headers that elasticsearch-py
+   negotiates from its own client major. Constructor-level ``headers=`` on
+   ``Elasticsearch.__init__`` and the ``elasticsearch_configs`` section do
+   **not** work for this purpose — elasticsearch-py re-applies its own
+   ``compatible-with=<client_major>`` headers right before the request goes
+   out, after any constructor headers.
+
 When the option is unset the client behaves as before (negotiating its own
 major version).
 
diff --git a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
index b55cd24823305..ca888db8a6f99 100644
--- a/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
+++ b/providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py
@@ -26,6 +26,7 @@
 
 from __future__ import annotations
 
+import functools
 import re
 from typing import TYPE_CHECKING
 
@@ -92,16 +93,23 @@ def apply_compat_with(client: elasticsearch.Elasticsearch) -> elasticsearch.Elas
     sub = rf"application/vnd.elasticsearch+\g<1>; compatible-with={compat}"
     original_perform_request = transport.perform_request
 
-    def perform_request(method, target, *, body=None, headers=None, **kwargs):  # type: ignore[no-untyped-def]
+    # Accept ``*args, **kwargs`` so the wrapper survives future elastic_transport
+    # ``perform_request`` signature changes (new keyword-only params, reordered
+    # positionals). ``functools.wraps`` preserves the original ``__name__`` /
+    # ``__doc__`` / ``__wrapped__`` for tooling and introspection.
+    @functools.wraps(original_perform_request)
+    def perform_request(*args, **kwargs):  # type: ignore[no-untyped-def]
+        headers = kwargs.get("headers")
         if not headers:
-            return original_perform_request(method, target, body=body, headers=headers, **kwargs)
+            return original_perform_request(*args, **kwargs)
         # Walk every key case-insensitively so a future elastic_transport that
         # forwards PascalCase headers does not silently bypass the rewrite.
         merged = dict(headers)
         for key in list(merged):
             if key.lower() in ("accept", "content-type") and merged[key]:
                 merged[key] = _COMPAT_MIMETYPE_RE.sub(sub, merged[key])
-        return original_perform_request(method, target, body=body, headers=merged, **kwargs)
+        kwargs["headers"] = merged
+        return original_perform_request(*args, **kwargs)
 
     # ``setattr`` instead of direct attribute assignment so mypy does not flag a
     # ``method-assign`` error — we are *intentionally* shadowing the bound method

From 64adc818294092b010ed1c3380b1a2c953822bc1 Mon Sep 17 00:00:00 2001
From: Peter <Peter.cheon@kakaocorp.com>
Date: Tue, 12 May 2026 10:50:19 +0900
Subject: [PATCH 7/7] Add 'misconfiguration' to spelling wordlist

Used in providers/elasticsearch/docs/logging/index.rst to describe the
fail-fast behavior when [elasticsearch] es_compat_with is set to an
invalid value. The wordlist already contained 'misconfigured' but the
noun form was missing, causing the --spellcheck-only docs build to fail.
---
 docs/spelling_wordlist.txt | 1 +
 1 file changed, 1 insertion(+)

diff --git a/docs/spelling_wordlist.txt b/docs/spelling_wordlist.txt
index b4460b21e7563..26662d523a3ff 100644
--- a/docs/spelling_wordlist.txt
+++ b/docs/spelling_wordlist.txt
@@ -1031,6 +1031,7 @@ milli
 millis
 milton
 minikube
+misconfiguration
 misconfigured
 Mixin
 mixin