fix(elasticsearch): pin compatible-with at the transport layer to keep ES 8 servers working

docs/spelling_wordlist.txt

@@ -1031,6 +1031,7 @@ milli
 millis
 milton
 minikube
+misconfiguration
 misconfigured
 Mixin
 mixin

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
 .....
 

providers/elasticsearch/docs/logging/index.rst

@@ -93,6 +93,48 @@ 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
+
+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).
+
 .. _elasticsearch-document-schema:
 
 Expected Elasticsearch document schema

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:

providers/elasticsearch/src/airflow/providers/elasticsearch/_compat.py

@@ -0,0 +1,119 @@
+#
+# 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
+
+import functools
+import re
+from typing import TYPE_CHECKING
+
+from airflow.providers.common.compat.sdk import AirflowConfigException, conf
+
+if TYPE_CHECKING:
+    import elasticsearch
+
+
+# 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:
+    """
+    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 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.
+    """
+    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__:
+        # Already wrapped on this transport instance — no-op so repeated calls
+        # to ``apply_compat_with`` (e.g. across hook reuse) stay idempotent.
+        return client
+
+    sub = rf"application/vnd.elasticsearch+\g<1>; compatible-with={compat}"
+    original_perform_request = transport.perform_request
+
+    # 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(*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])
+        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
+    # at the instance level (the idempotency guard above checks the instance
+    # ``__dict__``).
+    setattr(transport, "perform_request", perform_request)
+    return client

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": {

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,9 @@ 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 +248,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
 

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
@@ -651,7 +652,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)