Purge [elasticsearch_configs] use_ssl config param in ElasticsearchTaskHandler

airflow/providers/elasticsearch/log/es_task_handler.py

@@ -106,6 +106,15 @@ def __init__(
         # Read more at: https://elasticsearch-py.readthedocs.io/en/v8.8.2/api.html#module-elasticsearch
         if es_kwargs.get("retry_timeout"):
             es_kwargs["retry_on_timeout"] = es_kwargs.pop("retry_timeout")
+        # This parameter was removed in elasticsearch>8, however until Airflow 2.7.1 this parameter
+        # provided by default as False in `[elasticsearch_configs] config` section
+        use_ssl = es_kwargs.pop("use_ssl", None)
+        if use_ssl:  # If user set this config param explicitly to True we could warn
+            warnings.warn(
+                "Passing `[elasticsearch_configs] use_ssl` to ElasticsearchTaskHandler has no effect. "
+                "Please remove it from Airflow Configuration.",
+                UserWarning,
+            )
         host = self.format_url(host)
         super().__init__(base_log_folder, filename_template)
         self.closed = False

tests/providers/elasticsearch/log/test_es_task_handler.py

@@ -676,6 +676,42 @@ def test_dynamic_offset(self, stdout_mock, ti, time_machine):
         assert second_log["asctime"] == t2.format("YYYY-MM-DDTHH:mm:ss.SSSZZ")
         assert third_log["asctime"] == t3.format("YYYY-MM-DDTHH:mm:ss.SSSZZ")
 
+    def test_elasticsearch_constructor_use_ssl(self):
+        """Test `[elasticsearch_configs] use_ssl` default parameter prior Airflow 2.7.1"""
+        with mock.patch(
+            "airflow.providers.elasticsearch.log.es_task_handler.elasticsearch.Elasticsearch"
+        ) as mock_es:
+            es_kwargs = {"use_ssl": False, "legit_es8_parameter": "foo"}
+            ElasticsearchTaskHandler(
+                base_log_folder="dummy_folder",
+                end_of_log_mark="end_of_log_mark",
+                write_stdout=False,
+                json_format=False,
+                json_fields="fields",
+                host_field="host",
+                offset_field="offset",
+                es_kwargs=es_kwargs,
+            )
+            mock_es.assert_called_once_with(mock.ANY, legit_es8_parameter="foo")
+
+    def test_elasticsearch_constructor_use_ssl_warn_on_true(self):
+        with mock.patch(
+            "airflow.providers.elasticsearch.log.es_task_handler.elasticsearch.Elasticsearch"
+        ) as mock_es:
+            es_kwargs = {"use_ssl": True, "legit_es8_parameter": "bar"}
+            with pytest.warns(UserWarning, match="Passing `\[elasticsearch_configs\] use_ssl`"):
+                ElasticsearchTaskHandler(
+                    base_log_folder="dummy_folder",
+                    end_of_log_mark="end_of_log_mark",
+                    write_stdout=False,
+                    json_format=False,
+                    json_fields="fields",
+                    host_field="host",
+                    offset_field="offset",
+                    es_kwargs=es_kwargs,
+                )
+            mock_es.assert_called_once_with(mock.ANY, legit_es8_parameter="bar")
+
 
 def test_safe_attrgetter():
     class A: