This is an automated email from the ASF dual-hosted git repository.
eladkal pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 91dcbd9a7da Support elasticsearch 9 (#64070)
91dcbd9a7da is described below
commit 91dcbd9a7da9081c3ad45ce725d9973d8825ac46
Author: Subham <[email protected]>
AuthorDate: Tue Mar 24 00:28:17 2026 +0530
Support elasticsearch 9 (#64070)
* Support Elasticsearch 9 in Elasticsearch Provider
---
providers/elasticsearch/docs/index.rst | 2 +-
providers/elasticsearch/pyproject.toml | 2 +-
.../providers/elasticsearch/hooks/elasticsearch.py | 6 +++---
.../elasticsearch/log/test_es_remote_log_io.py | 2 +-
.../unit/elasticsearch/hooks/test_elasticsearch.py | 8 +++-----
.../log/elasticmock/fake_elasticsearch.py | 6 ++++--
.../unit/elasticsearch/log/test_es_task_handler.py | 18 +++++++++---------
uv.lock | 4 ++--
8 files changed, 24 insertions(+), 24 deletions(-)
diff --git a/providers/elasticsearch/docs/index.rst
b/providers/elasticsearch/docs/index.rst
index 4182094239c..2436753f5e0 100644
--- a/providers/elasticsearch/docs/index.rst
+++ b/providers/elasticsearch/docs/index.rst
@@ -105,7 +105,7 @@ PIP package Version required
``apache-airflow`` ``>=2.11.0``
``apache-airflow-providers-common-compat`` ``>=1.12.0``
``apache-airflow-providers-common-sql`` ``>=1.32.0``
-``elasticsearch`` ``>=8.10,<9``
+``elasticsearch`` ``<10,>=8.10``
========================================== ==================
Cross provider package dependencies
diff --git a/providers/elasticsearch/pyproject.toml
b/providers/elasticsearch/pyproject.toml
index cbfb6576ade..bf22e075e58 100644
--- a/providers/elasticsearch/pyproject.toml
+++ b/providers/elasticsearch/pyproject.toml
@@ -62,7 +62,7 @@ dependencies = [
"apache-airflow>=2.11.0",
"apache-airflow-providers-common-compat>=1.12.0",
"apache-airflow-providers-common-sql>=1.32.0",
- "elasticsearch>=8.10,<9",
+ "elasticsearch>=8.10,<10",
]
[dependency-groups]
diff --git
a/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
b/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
index 2bbc5bab176..e8f0cfad00c 100644
---
a/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
+++
b/providers/elasticsearch/src/airflow/providers/elasticsearch/hooks/elasticsearch.py
@@ -86,7 +86,7 @@ class ElasticsearchSQLCursor:
self.body["query"] = statement
if params:
self.body["params"] = params
- self.response = self.es.sql.query(body=self.body)
+ self.response = self.es.sql.query(**self.body)
if self.cursor:
self.body["cursor"] = self.cursor
else:
@@ -135,7 +135,7 @@ class ESConnection:
netloc = f"{host}:{port}"
self.url = parse.urlunparse((scheme, netloc, "/", None, None, None))
if user and password:
- self.es = Elasticsearch(self.url, http_auth=(user, password),
**kwargs)
+ self.es = Elasticsearch(self.url, basic_auth=(user, password),
**kwargs)
else:
self.es = Elasticsearch(self.url, **kwargs)
@@ -266,5 +266,5 @@ class ElasticsearchPythonHook(BaseHook):
:returns: dict: The response 'hits' object from Elasticsearch
"""
es_client = self.get_conn
- result = es_client.search(index=index, body=query)
+ result = es_client.search(index=index, **query)
return result["hits"]
diff --git
a/providers/elasticsearch/tests/integration/elasticsearch/log/test_es_remote_log_io.py
b/providers/elasticsearch/tests/integration/elasticsearch/log/test_es_remote_log_io.py
index 89a12a3a39b..e6da3dc7514 100644
---
a/providers/elasticsearch/tests/integration/elasticsearch/log/test_es_remote_log_io.py
+++
b/providers/elasticsearch/tests/integration/elasticsearch/log/test_es_remote_log_io.py
@@ -136,7 +136,7 @@ class TestElasticsearchRemoteLogIOIntegration:
"offset": 1,
"error_detail": error_detail,
}
- self.elasticsearch_io.client.index(index=self.target_index, body=body)
+ self.elasticsearch_io.client.index(index=self.target_index,
document=body)
self.elasticsearch_io.client.indices.refresh(index=self.target_index)
log_source_info, log_messages = self.elasticsearch_io.read("", ti)
diff --git
a/providers/elasticsearch/tests/unit/elasticsearch/hooks/test_elasticsearch.py
b/providers/elasticsearch/tests/unit/elasticsearch/hooks/test_elasticsearch.py
index 78c141a9baa..ed9056ae350 100644
---
a/providers/elasticsearch/tests/unit/elasticsearch/hooks/test_elasticsearch.py
+++
b/providers/elasticsearch/tests/unit/elasticsearch/hooks/test_elasticsearch.py
@@ -236,11 +236,9 @@ class TestElasticsearchSQLHook:
es_connection = ESConnection(host="localhost", port=9200)
response = es_connection.execute_sql("SELECT * FROM hollywood.actors")
mock_es_sql_client.query.assert_called_once_with(
- body={
- "fetch_size": 1000,
- "field_multi_value_leniency": False,
- "query": "SELECT * FROM hollywood.actors",
- }
+ fetch_size=1000,
+ field_multi_value_leniency=False,
+ query="SELECT * FROM hollywood.actors",
)
assert response == RESPONSE_WITHOUT_CURSOR
diff --git
a/providers/elasticsearch/tests/unit/elasticsearch/log/elasticmock/fake_elasticsearch.py
b/providers/elasticsearch/tests/unit/elasticsearch/log/elasticmock/fake_elasticsearch.py
index df101f2f918..4b9d81c0210 100644
---
a/providers/elasticsearch/tests/unit/elasticsearch/log/elasticmock/fake_elasticsearch.py
+++
b/providers/elasticsearch/tests/unit/elasticsearch/log/elasticmock/fake_elasticsearch.py
@@ -316,7 +316,9 @@ class FakeElasticsearch(Elasticsearch):
"version",
"version_type",
)
- def index(self, index, doc_type, body, id=None, params=None, headers=None):
+ def index(self, index, document=None, doc_type=None, body=None, id=None,
params=None, headers=None):
+ if document is None:
+ document = body
if index not in self.__documents_dict:
self.__documents_dict[index] = []
@@ -329,7 +331,7 @@ class FakeElasticsearch(Elasticsearch):
{
"_type": doc_type,
"_id": id,
- "_source": body,
+ "_source": document,
"_index": index,
"_version": version,
"_headers": headers,
diff --git
a/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
index 8d6a0ba616c..c679eedf38e 100644
---
a/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
+++
b/providers/elasticsearch/tests/unit/elasticsearch/log/test_es_task_handler.py
@@ -139,7 +139,7 @@ class TestElasticsearchTaskHandler:
"offset": 1,
"event": self.test_message,
}
- self.es.index(index=self.index_name, doc_type=self.doc_type,
body=self.body, id=1)
+ self.es.index(index=self.index_name, doc_type=self.doc_type,
document=self.body, id=1)
def teardown_method(self):
shutil.rmtree(self.local_log_location.split(os.path.sep)[0],
ignore_errors=True)
@@ -375,7 +375,7 @@ class TestElasticsearchTaskHandler:
"log_id": similar_log_id,
"offset": 1,
}
- self.es.index(index=self.index_name, doc_type=self.doc_type,
body=another_body, id=1)
+ self.es.index(index=self.index_name, doc_type=self.doc_type,
document=another_body, id=1)
ts = pendulum.now()
logs, metadatas = self.es_task_handler.read(
@@ -624,7 +624,7 @@ class TestElasticsearchTaskHandler:
"levelname": "INFO",
}
self.es_task_handler.set_context(ti)
- self.es.index(index=self.index_name, doc_type=self.doc_type,
body=self.body, id=id)
+ self.es.index(index=self.index_name, doc_type=self.doc_type,
document=self.body, id=id)
logs, _ = self.es_task_handler.read(
ti, 1, {"offset": 0, "last_log_timestamp": str(ts), "end_of_log":
False}
@@ -658,7 +658,7 @@ class TestElasticsearchTaskHandler:
"levelname": "INFO",
}
self.es_task_handler.set_context(ti)
- self.es.index(index=self.index_name, doc_type=self.doc_type,
body=self.body, id=id)
+ self.es.index(index=self.index_name, doc_type=self.doc_type,
document=self.body, id=id)
logs, _ = self.es_task_handler.read(
ti, 1, {"offset": 0, "last_log_timestamp": str(ts), "end_of_log":
False}
@@ -685,7 +685,7 @@ class TestElasticsearchTaskHandler:
"log": {"offset": 1},
"host": {"name": "somehostname"},
}
- self.es.index(index=self.index_name, doc_type=self.doc_type,
body=self.body, id=id)
+ self.es.index(index=self.index_name, doc_type=self.doc_type,
document=self.body, id=id)
logs, _ = self.es_task_handler.read(
ti, 1, {"offset": 0, "last_log_timestamp": str(ts), "end_of_log":
False}
@@ -925,14 +925,14 @@ def test_retrieve_config_keys():
with conf_vars(
{
("elasticsearch_configs", "http_compress"): "False",
- ("elasticsearch_configs", "timeout"): "10",
+ ("elasticsearch_configs", "request_timeout"): "10",
}
):
args_from_config = get_es_kwargs_from_config().keys()
# verify_certs comes from default config value
assert "verify_certs" in args_from_config
- # timeout comes from config provided value
- assert "timeout" in args_from_config
+ # request_timeout comes from config provided value
+ assert "request_timeout" in args_from_config
# http_compress comes from config value
assert "http_compress" in args_from_config
assert "self" not in args_from_config
@@ -1290,7 +1290,7 @@ class TestBuildStructuredLogFields:
}
],
}
- es.index(index="test_index", doc_type="log", body=body, id=1)
+ es.index(index="test_index", doc_type="log", document=body, id=1)
# Patch the IO layer to return our fake document
mock_hit_dict = body.copy()
diff --git a/uv.lock b/uv.lock
index 0d75a8127c2..823473380e8 100644
--- a/uv.lock
+++ b/uv.lock
@@ -1,5 +1,5 @@
version = 1
-revision = 3
+revision = 2
requires-python = ">=3.10"
resolution-markers = [
"python_full_version >= '3.15' and sys_platform == 'win32'",
@@ -4572,7 +4572,7 @@ requires-dist = [
{ name = "apache-airflow", editable = "." },
{ name = "apache-airflow-providers-common-compat", editable =
"providers/common/compat" },
{ name = "apache-airflow-providers-common-sql", editable =
"providers/common/sql" },
- { name = "elasticsearch", specifier = ">=8.10,<9" },
+ { name = "elasticsearch", specifier = ">=8.10,<10" },
]
[package.metadata.requires-dev]
