diff --git a/airflow/providers/amazon/aws/log/s3_task_handler.py b/airflow/providers/amazon/aws/log/s3_task_handler.py
index fec1344c76c..a1ba2d57ef8 100644
--- a/airflow/providers/amazon/aws/log/s3_task_handler.py
+++ b/airflow/providers/amazon/aws/log/s3_task_handler.py
@@ -62,12 +62,7 @@ def hook(self):
         )
 
     def set_context(self, ti: TaskInstance, *, identifier: str | None = None) -> None:
-        # todo: remove-at-min-airflow-version-2.8
-        #   after Airflow 2.8 can always pass `identifier`
-        if getattr(super(), "supports_task_context_logging", False):
-            super().set_context(ti, identifier=identifier)
-        else:
-            super().set_context(ti)
+        super().set_context(ti, identifier=identifier)
         # Local location and remote location is needed to open and
         # upload local log file to S3 remote storage.
         if TYPE_CHECKING:
diff --git a/airflow/providers/elasticsearch/log/es_task_handler.py b/airflow/providers/elasticsearch/log/es_task_handler.py
index 7eda572c54c..6fd16ca712a 100644
--- a/airflow/providers/elasticsearch/log/es_task_handler.py
+++ b/airflow/providers/elasticsearch/log/es_task_handler.py
@@ -468,12 +468,7 @@ def set_context(self, ti: TaskInstance, *, identifier: str | None = None) -> Non
             self.handler.setLevel(self.level)
             self.handler.setFormatter(self.formatter)
         else:
-            # todo: remove-at-min-airflow-version-2.8
-            #   after Airflow 2.8 can always pass `identifier`
-            if getattr(super(), "supports_task_context_logging", False):
-                super().set_context(ti, identifier=identifier)
-            else:
-                super().set_context(ti)
+            super().set_context(ti, identifier=identifier)
         self.context_set = True
 
     def close(self) -> None:
diff --git a/airflow/providers/google/cloud/log/gcs_task_handler.py b/airflow/providers/google/cloud/log/gcs_task_handler.py
index 8cdde520e46..8c11b2692e6 100644
--- a/airflow/providers/google/cloud/log/gcs_task_handler.py
+++ b/airflow/providers/google/cloud/log/gcs_task_handler.py
@@ -129,12 +129,7 @@ def client(self) -> storage.Client:
         )
 
     def set_context(self, ti: TaskInstance, *, identifier: str | None = None) -> None:
-        # todo: remove-at-min-airflow-version-2.8
-        #   after Airflow 2.8 can always pass `identifier`
-        if getattr(super(), "supports_task_context_logging", False):
-            super().set_context(ti, identifier=identifier)
-        else:
-            super().set_context(ti)
+        super().set_context(ti, identifier=identifier)
         # Log relative path is used to construct local and remote
         # log path to upload log files into GCS and read from the
         # remote location.
diff --git a/airflow/providers/microsoft/azure/log/wasb_task_handler.py b/airflow/providers/microsoft/azure/log/wasb_task_handler.py
index df21930237f..d32a3463d74 100644
--- a/airflow/providers/microsoft/azure/log/wasb_task_handler.py
+++ b/airflow/providers/microsoft/azure/log/wasb_task_handler.py
@@ -81,12 +81,7 @@ def hook(self):
             return None
 
     def set_context(self, ti: TaskInstance, *, identifier: str | None = None) -> None:
-        # todo: remove-at-min-airflow-version-2.8
-        #   after Airflow 2.8 can always pass `identifier`
-        if getattr(super(), "supports_task_context_logging", False):
-            super().set_context(ti, identifier=identifier)
-        else:
-            super().set_context(ti)
+        super().set_context(ti, identifier=identifier)
         # Local location and remote location is needed to open and
         # upload local log file to Wasb remote storage.
         if TYPE_CHECKING:
diff --git a/airflow/providers/opensearch/log/os_task_handler.py b/airflow/providers/opensearch/log/os_task_handler.py
index 99f9d416988..11865b2b4ff 100644
--- a/airflow/providers/opensearch/log/os_task_handler.py
+++ b/airflow/providers/opensearch/log/os_task_handler.py
@@ -215,12 +215,7 @@ def set_context(self, ti: TaskInstance, *, identifier: str | None = None) -> Non
             self.handler.setLevel(self.level)
             self.handler.setFormatter(self.formatter)
         else:
-            # todo: remove-at-min-airflow-version-2.8
-            #   after Airflow 2.8 can always pass `identifier`
-            if getattr(super(), "supports_task_context_logging", False):
-                super().set_context(ti, identifier=identifier)
-            else:
-                super().set_context(ti)
+            super().set_context(ti, identifier=identifier)
         self.context_set = True
 
     def emit(self, record):