Repository: apache/airflow

Issue:
`AirbyteTriggerSyncOperator` does not work on deferrable mode
### Apache Airflow Provider(s)

airbyte

### Versions of Apache Airflow Providers

apache-airflow-providers-airbyte==3.6.0
apache-airflow-providers-http==4.5.1

### Apache Airflow version

2.7.1

### Operating System

Ubuntu 22.04.3 LTS

### Deployment

Docker-Compose

### Deployment details

_No response_

### What happened

`AirbyteTriggerSyncOperator` does not work properly in deferrable mode.

### What you think should happen instead

_No response_

### How to reproduce

Create the following DAG, replacing `"id_of_airbyte_connection_to_be_synced"` with the corresponding value.
```python
from datetime import datetime

from airflow import DAG
from airflow.providers.airbyte.operators.airbyte import \
    AirbyteTriggerSyncOperator


AIRFLOW_AIRBYTE_CONN_ID = "airbyte_default"
AIRBYTE_CONNECTION_ID = "id_of_airbyte_connection_to_be_synced"  # replace this


with DAG("test_dag", start_date=datetime.min, catchup=False) as dag:
    not_deferrable = AirbyteTriggerSyncOperator(
        task_id="not_deferrable",
        airbyte_conn_id=AIRFLOW_AIRBYTE_CONN_ID,
        connection_id=AIRBYTE_CONNECTION_ID,
        deferrable=False
    )

    deferrable = AirbyteTriggerSyncOperator(
        task_id="deferrable",
        airbyte_conn_id=AIRFLOW_AIRBYTE_CONN_ID,
        connection_id=AIRBYTE_CONNECTION_ID,
        deferrable=True
    )

```
The `not_deferrable` task works, while the `deferrable` task fails. The `"airbyte_default"` connection is set in all containers via the environment variable
```
AIRFLOW_CONN_AIRBYTE_DEFAULT='{
    "conn_type": "airbyte",
    "host": "airbyte-proxy",
    "port": 8000
}'
```

### Anything else

The deferrable task fails with the following log:
<details><summary>deferrable.log</summary>

```
78f82177fe4f
*** Found local files:
***   * /opt/airflow/logs/dag_id=test_dag/run_id=scheduled__2024-02-14T15:14:37.364888+00:00/task_id=deferrable/attempt=1.log
***   * /opt/airflow/logs/dag_id=test_dag/run_id=scheduled__2024-02-14T15:14:37.364888+00:00/task_id=deferrable/attempt=1.log.trigger.1205.log
[2024-02-15, 15:14:38 UTC] {taskinstance.py:1157} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: test_dag.deferrable scheduled__2024-02-14T15:14:37.364888+00:00 [queued]>
[2024-02-15, 15:14:38 UTC] {taskinstance.py:1157} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: test_dag.deferrable scheduled__2024-02-14T15:14:37.364888+00:00 [queued]>
[2024-02-15, 15:14:38 UTC] {taskinstance.py:1359} INFO - Starting attempt 1 of 1
[2024-02-15, 15:14:38 UTC] {taskinstance.py:1380} INFO - Executing <Task(AirbyteTriggerSyncOperator): deferrable> on 2024-02-14 15:14:37.364888+00:00
[2024-02-15, 15:14:38 UTC] {standard_task_runner.py:57} INFO - Started process 7210 to run task
[2024-02-15, 15:14:38 UTC] {standard_task_runner.py:84} INFO - Running: ['***', 'tasks', 'run', 'test_dag', 'deferrable', 'scheduled__2024-02-14T15:14:37.364888+00:00', '--job-id', '1213', '--raw', '--subdir', 'DAGS_FOLDER/test_dag.py', '--cfg-path', '/var/tmp/tmp7v9zx5oh']
[2024-02-15, 15:14:38 UTC] {standard_task_runner.py:85} INFO - Job 1213: Subtask deferrable
[2024-02-15, 15:14:38 UTC] {task_command.py:415} INFO - Running <TaskInstance: test_dag.deferrable scheduled__2024-02-14T15:14:37.364888+00:00 [running]> on host 78f82177fe4f
[2024-02-15, 15:14:38 UTC] {taskinstance.py:1660} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='***' AIRFLOW_CTX_DAG_ID='test_dag' AIRFLOW_CTX_TASK_ID='deferrable' AIRFLOW_CTX_EXECUTION_DATE='2024-02-14T15:14:37.364888+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='scheduled__2024-02-14T15:14:37.364888+00:00'
[2024-02-15, 15:14:38 UTC] {base.py:73} INFO - Using connection ID 'airbyte_default' for task execution.
[2024-02-15, 15:14:43 UTC] {airbyte.py:86} INFO - Job 224 was submitted to Airbyte Server
[2024-02-15, 15:14:43 UTC] {airbyte.py:88} INFO - Waiting for job 224 to complete
[2024-02-15, 15:14:43 UTC] {taskinstance.py:1524} INFO - Pausing task as DEFERRED. dag_id=test_dag, task_id=deferrable, execution_date=20240214T151437, start_date=20240215T151438
[2024-02-15, 15:14:44 UTC] {local_task_job_runner.py:225} INFO - Task exited with return code 100 (task deferral)
[2024-02-15, 15:14:44 UTC] {airbyte.py:104} INFO - Getting the status of job run 224.
[2024-02-15, 15:14:44 UTC] {base.py:73} INFO - Using connection ID 'airbyte_default' for task execution.
[2024-02-15, 15:14:44 UTC] {airbyte.py:88} INFO - URL for api request: airbyte-proxy/api/v1/jobs/get
[2024-02-15, 15:14:44 UTC] {triggerer_job_runner.py:599} INFO - Trigger test_dag/scheduled__2024-02-14T15:14:37.364888+00:00/deferrable/-1/1 (ID 17) fired: TriggerEvent<{'status': 'error', 'message': 'airbyte-proxy/api/v1/jobs/get', 'job_id': 224}>
[2024-02-15, 15:14:46 UTC] {taskinstance.py:1157} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: test_dag.deferrable scheduled__2024-02-14T15:14:37.364888+00:00 [queued]>
[2024-02-15, 15:14:46 UTC] {taskinstance.py:1157} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: test_dag.deferrable scheduled__2024-02-14T15:14:37.364888+00:00 [queued]>
[2024-02-15, 15:14:46 UTC] {taskinstance.py:1357} INFO - Resuming after deferral
[2024-02-15, 15:14:46 UTC] {taskinstance.py:1380} INFO - Executing <Task(AirbyteTriggerSyncOperator): deferrable> on 2024-02-14 15:14:37.364888+00:00
[2024-02-15, 15:14:46 UTC] {standard_task_runner.py:57} INFO - Started process 7238 to run task
[2024-02-15, 15:14:46 UTC] {standard_task_runner.py:84} INFO - Running: ['***', 'tasks', 'run', 'test_dag', 'deferrable', 'scheduled__2024-02-14T15:14:37.364888+00:00', '--job-id', '1214', '--raw', '--subdir', 'DAGS_FOLDER/test_dag.py', '--cfg-path', '/var/tmp/tmp7eh43099']
[2024-02-15, 15:14:46 UTC] {standard_task_runner.py:85} INFO - Job 1214: Subtask deferrable
[2024-02-15, 15:14:46 UTC] {task_command.py:415} INFO - Running <TaskInstance: test_dag.deferrable scheduled__2024-02-14T15:14:37.364888+00:00 [running]> on host 78f82177fe4f
[2024-02-15, 15:14:46 UTC] {taskinstance.py:1935} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/models/baseoperator.py", line 1608, in resume_execution
    return execute_callable(context)
           ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/airflow/.local/lib/python3.11/site-packages/airflow/providers/airbyte/operators/airbyte.py", line 124, in execute_complete
    raise AirflowException(event["message"])
airflow.exceptions.AirflowException: airbyte-proxy/api/v1/jobs/get
[2024-02-15, 15:14:46 UTC] {taskinstance.py:1398} INFO - Marking task as FAILED. dag_id=test_dag, task_id=deferrable, execution_date=20240214T151437, start_date=20240215T151438, end_date=20240215T151446
[2024-02-15, 15:14:46 UTC] {standard_task_runner.py:104} ERROR - Failed to execute job 1214 for task deferrable (airbyte-proxy/api/v1/jobs/get; 7238)
[2024-02-15, 15:14:46 UTC] {local_task_job_runner.py:228} INFO - Task exited with return code 1
[2024-02-15, 15:14:46 UTC] {taskinstance.py:2776} INFO - 0 downstream tasks scheduled from follow-on schedule check
```

</details>

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)


Discussion:
boring-cyborg[bot]: Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide (https://github.com/apache/airflow/blob/main/contributing-docs/README.rst)
Here are some useful points:
- Pay attention to the quality of your code (ruff, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/dev/breeze/doc/README.rst) for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#coding-style-and-best-practices).
- Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
Apache Airflow is a community-driven project and together we are making it better 🚀.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack

josix: Hi @hussein-awala thanks for the advise, I've added the test for it. PTAL, thanks.

Patch:
diff --git a/airflow/api_connexion/endpoints/import_error_endpoint.py b/airflow/api_connexion/endpoints/import_error_endpoint.py
index a56cb97c41..274d842d18 100644
--- a/airflow/api_connexion/endpoints/import_error_endpoint.py
+++ b/airflow/api_connexion/endpoints/import_error_endpoint.py
@@ -94,11 +94,9 @@ def get_import_errors(
     if not can_read_all_dags:
         # if the user doesn't have access to all DAGs, only display errors from visible DAGs
         readable_dag_ids = security.get_readable_dags()
-        dagfiles_subq = (
-            select(DagModel.fileloc).distinct().where(DagModel.dag_id.in_(readable_dag_ids)).subquery()
-        )
-        query = query.where(ImportErrorModel.filename.in_(dagfiles_subq))
-        count_query = count_query.where(ImportErrorModel.filename.in_(dagfiles_subq))
+        dagfiles_stmt = select(DagModel.fileloc).distinct().where(DagModel.dag_id.in_(readable_dag_ids))
+        query = query.where(ImportErrorModel.filename.in_(dagfiles_stmt))
+        count_query = count_query.where(ImportErrorModel.filename.in_(dagfiles_stmt))
 
     total_entries = session.scalars(count_query).one()
     import_errors = session.scalars(query.offset(offset).limit(limit)).all()
diff --git a/airflow/providers/airbyte/hooks/airbyte.py b/airflow/providers/airbyte/hooks/airbyte.py
index 7c1132fa57..1cecb5b031 100644
--- a/airflow/providers/airbyte/hooks/airbyte.py
+++ b/airflow/providers/airbyte/hooks/airbyte.py
@@ -71,7 +71,12 @@ class AirbyteHook(HttpHook):
     async def get_headers_tenants_from_connection(self) -> tuple[dict[str, Any], str]:
         """Get Headers, tenants from the connection details."""
         connection: Connection = await sync_to_async(self.get_connection)(self.http_conn_id)
-        base_url = connection.host
+        # schema defaults to HTTP
+        schema = connection.schema if connection.schema else "http"
+        base_url = f"{schema}://{connection.host}"
+
+        if connection.port:
+            base_url += f":{connection.port}"
 
         if self.api_type == "config":
             credentials = f"{connection.login}:{connection.password}"
diff --git a/airflow/utils/db_cleanup.py b/airflow/utils/db_cleanup.py
index e119c4228b..4475209b77 100644
--- a/airflow/utils/db_cleanup.py
+++ b/airflow/utils/db_cleanup.py
@@ -186,9 +186,7 @@ def _do_delete(*, query, orm_model, skip_archive, session):
     if dialect_name == "sqlite":
         pk_cols = source_table.primary_key.columns
         delete = source_table.delete().where(
-            tuple_(*pk_cols).in_(
-                select(*[target_table.c[x.name] for x in source_table.primary_key.columns]).subquery()
-            )
+            tuple_(*pk_cols).in_(select(*[target_table.c[x.name] for x in source_table.primary_key.columns]))
         )
     else:
         delete = source_table.delete().where(
diff --git a/airflow/www/static/js/api/useGridData.ts b/airflow/www/static/js/api/useGridData.ts
index ee9325e852..1a633acea3 100644
--- a/airflow/www/static/js/api/useGridData.ts
+++ b/airflow/www/static/js/api/useGridData.ts
@@ -33,9 +33,9 @@ import useFilters, {
   FILTER_UPSTREAM_PARAM,
   ROOT_PARAM,
 } from "src/dag/useFilters";
-import type { Task, DagRun, RunOrdering } from "src/types";
+import type { Task, DagRun, RunOrdering, API } from "src/types";
 import { camelCase } from "lodash";
-import useSelection, { RUN_ID } from "src/dag/useSelection";
+import useSelection from "src/dag/useSelection";
 
 const DAG_ID_PARAM = "dag_id";
 
@@ -80,9 +80,12 @@ const useGridData = () => {
       filterDownstream,
       filterUpstream,
     },
+    onBaseDateChange,
   } = useFilters();
-  const { firstRunIdSetByUrl } = useSelection();
-
+  const {
+    onSelect,
+    selected: { taskId, runId },
+  } = useSelection();
   const query = useQuery(
     [
       "gridData",
@@ -93,7 +96,7 @@ const useGridData = () => {
       root,
       filterUpstream,
       filterDownstream,
-      firstRunIdSetByUrl,
+      runId,
     ],
     async () => {
       const params = {
@@ -105,11 +108,28 @@ const useGridData = () => {
         [NUM_RUNS_PARAM]: numRuns,
         [RUN_TYPE_PARAM]: runType,
         [RUN_STATE_PARAM]: runState,
-        [RUN_ID]: firstRunIdSetByUrl || "",
       };
       const response = await axios.get<AxiosResponse, GridData>(gridDataUrl, {
         params,
       });
+      if (runId && !response.dagRuns.find((dr) => dr.runId === runId)) {
+        const dagRunUrl = getMetaValue("dag_run_url")
+          .replace("__DAG_ID__", dagId)
+          .replace("__DAG_RUN_ID__", runId);
+
+        // If the run id cannot be found in the response, try fetching it to see if its real and then adjust the base date filter
+        try {
+          const selectedRun = await axios.get<AxiosResponse, API.DAGRun>(
+            dagRunUrl
+          );
+          if (selectedRun?.executionDate) {
+            onBaseDateChange(selectedRun.executionDate);
+          }
+          // otherwise the run_id isn't valid and we should unselect it
+        } catch (e) {
+          onSelect({ taskId });
+        }
+      }
       // turn off auto refresh if there are no active runs
       if (!areActiveRuns(response.dagRuns)) stopRefresh();
       return response;
diff --git a/airflow/www/static/js/dag/Main.tsx b/airflow/www/static/js/dag/Main.tsx
index 2d70b452d5..144aba94a9 100644
--- a/airflow/www/static/js/dag/Main.tsx
+++ b/airflow/www/static/js/dag/Main.tsx
@@ -47,7 +47,6 @@ import FilterBar from "./nav/FilterBar";
 import LegendRow from "./nav/LegendRow";
 import useToggleGroups from "./useToggleGroups";
 import keyboardShortcutIdentifier from "./keyboardShortcutIdentifier";
-import { DagRunSelectionContext, RUN_ID } from "./useSelection";
 
 const detailsPanelKey = "hideDetailsPanel";
 const minPanelWidth = 300;
@@ -74,7 +73,7 @@ const headerHeight =
     10
   ) || 0;
 
-const MainInContext = () => {
+const Main = () => {
   const {
     data: { groups },
     isLoading,
@@ -319,15 +318,4 @@ const MainInContext = () => {
   );
 };
 
-const Main = () => {
-  const [searchParams] = useSearchParams();
-  const [firstRunIdSetByUrl] = useState(searchParams.get(RUN_ID));
-
-  return (
-    <DagRunSelectionContext.Provider value={firstRunIdSetByUrl}>
-      <MainInContext />
-    </DagRunSelectionContext.Provider>
-  );
-};
-
 export default Main;
diff --git a/airflow/www/static/js/dag/details/Header.tsx b/airflow/www/static/js/dag/details/Header.tsx
index e1222270d4..d13dfa6c81 100644
--- a/airflow/www/static/js/dag/details/Header.tsx
+++ b/airflow/www/static/js/dag/details/Header.tsx
@@ -51,17 +51,14 @@ const Header = ({ mapIndex }: Props) => {
   } = useSelection();
 
   const dagRun = dagRuns.find((r) => r.runId === runId);
-
   const group = getTask({ taskId, task: groups });
 
-  // If runId and/or taskId can't be found remove the selection
+  // If taskId can't be found remove the selection
   useEffect(() => {
-    if (runId && !dagRun && taskId && !group) {
-      clearSelection();
-    } else if (runId && !dagRun) {
-      onSelect({ taskId });
+    if (taskId && !group) {
+      onSelect({ runId });
     }
-  }, [dagRun, taskId, group, runId, onSelect, clearSelection]);
+  }, [taskId, group, onSelect, runId]);
 
   let runLabel;
   if (dagRun && runId) {
diff --git a/airflow/www/static/js/dag/useSelection.ts b/airflow/www/static/js/dag/useSelection.ts
index 00ead40c52..04113a870c 100644
--- a/airflow/www/static/js/dag/useSelection.ts
+++ b/airflow/www/static/js/dag/useSelection.ts
@@ -17,7 +17,6 @@
  * under the License.
  */
 
-import { createContext, useContext } from "react";
 import { useSearchParams } from "react-router-dom";
 import {
   LIMIT_PARAM,
@@ -35,13 +34,8 @@ export interface SelectionProps {
   mapIndex?: number;
 }
 
-// The first run_id need to be treated differently from the selection, because it is used in backend to
-// calculate the base_date, which we don't want jumping around when user is clicking in the grid.
-export const DagRunSelectionContext = createContext<string | null>(null);
-
 const useSelection = () => {
   const [searchParams, setSearchParams] = useSearchParams();
-  const firstRunIdSetByUrl = useContext(DagRunSelectionContext);
 
   // Clear selection, but keep other search params
   const clearSelection = () => {
@@ -99,7 +93,6 @@ const useSelection = () => {
     },
     clearSelection,
     onSelect,
-    firstRunIdSetByUrl,
   };
 };
 
diff --git a/airflow/www/templates/airflow/dag.html b/airflow/www/templates/airflow/dag.html
index 56bfc27070..d9ef3c139a 100644
--- a/airflow/www/templates/airflow/dag.html
+++ b/airflow/www/templates/airflow/dag.html
@@ -84,6 +84,7 @@
   <meta name="datasets_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dataset_endpoint_get_datasets') }}">
   <meta name="event_logs_api" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_event_log_endpoint_get_event_logs') }}">
   <meta name="audit_log_url" content="{{ url_for('LogModelView.list') }}">
+  <meta name="dag_run_url" content="{{ url_for('/api/v1.airflow_api_connexion_endpoints_dag_run_endpoint_get_dag_run', dag_id='__DAG_ID__', dag_run_id='__DAG_RUN_ID__') }}">
 
   <!-- End Urls -->
   <meta name="is_paused" content="{{ dag_is_paused }}">
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 8a5ea38270..ce0727ab4d 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -952,10 +952,7 @@ class Airflow(AirflowBaseView):
                     # if the user doesn't have access to all DAGs, only display errors from visible DAGs
                     import_errors = import_errors.where(
                         errors.ImportError.filename.in_(
-                            select(DagModel.fileloc)
-                            .distinct()
-                            .where(DagModel.dag_id.in_(filter_dag_ids))
-                            .subquery()
+                            select(DagModel.fileloc).distinct().where(DagModel.dag_id.in_(filter_dag_ids))
                         )
                     )
 
@@ -3170,7 +3167,6 @@ class Airflow(AirflowBaseView):
     def grid_data(self):
         """Return grid data."""
         dag_id = request.args.get("dag_id")
-        run_id = request.args.get("dag_run_id")
         dag = get_airflow_app().dag_bag.get_dag(dag_id)
 
         if not dag:
@@ -3188,43 +3184,25 @@ class Airflow(AirflowBaseView):
         if num_runs is None:
             num_runs = conf.getint("webserver", "default_dag_run_display_number")
 
-        dagrun = None
-        if run_id:
-            with create_session() as session:
-                dagrun = dag.get_dagrun(run_id=run_id, session=session)
-                if not dagrun:
-                    return {"error": f"can't find dag_run_id={run_id}"}, 404
-            base_date = dagrun.execution_date
-        else:
-            try:
-                base_date = timezone.parse(request.args["base_date"], strict=True)
-            except (KeyError, ValueError):
-                base_date = dag.get_latest_execution_date() or timezone.utcnow()
+        try:
+            base_date = timezone.parse(request.args["base_date"], strict=True)
+        except (KeyError, ValueError):
+            base_date = dag.get_latest_execution_date() or timezone.utcnow()
 
         with create_session() as session:
             query = select(DagRun).where(DagRun.dag_id == dag.dag_id, DagRun.execution_date <= base_date)
 
         run_types = request.args.getlist("run_type")
         if run_types:
-            if run_id:
-                return {"error": "Can not provide filters when dag_run_id filter is selected."}, 400
             query = query.where(DagRun.run_type.in_(run_types))
 
         run_states = request.args.getlist("run_state")
         if run_states:
-            if run_id:
-                return {"error": "Can not provide filters when dag_run_id filter is selected."}, 400
             query = query.where(DagRun.state.in_(run_states))
 
         dag_runs = wwwutils.sorted_dag_runs(
             query, ordering=dag.timetable.run_ordering, limit=num_runs, session=session
         )
-        if dagrun:
-            found_requested_run_id = any(True for d in dag_runs if d.run_id == run_id)
-            if not found_requested_run_id:
-                return {
-                    "error": f"Dag with dag_run_id={run_id} found, but not in selected time range or filters."
-                }, 404
 
         encoded_runs = [wwwutils.encode_dag_run(dr, json_encoder=utils_json.WebEncoder) for dr in dag_runs]
         data = {
diff --git a/docs/apache-airflow/howto/docker-compose/docker-compose.yaml b/docs/apache-airflow/howto/docker-compose/docker-compose.yaml
index 9effc2d504..92b0eab937 100644
--- a/docs/apache-airflow/howto/docker-compose/docker-compose.yaml
+++ b/docs/apache-airflow/howto/docker-compose/docker-compose.yaml
@@ -102,7 +102,9 @@ services:
     restart: always
 
   redis:
-    image: redis:latest
+    # Redis is limited to 7.2-bookworm due to licencing change
+    # https://redis.io/blog/redis-adopts-dual-source-available-licensing/
+    image: redis:7.2-bookworm
     expose:
       - 6379
     healthcheck:
diff --git a/pyproject.toml b/pyproject.toml
index 985acc75fa..ee55d35bc3 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -467,6 +467,7 @@ filterwarnings = [
     "error::pytest.PytestReturnNotNoneWarning",
     # Avoid building cartesian product which might impact performance
     "error:SELECT statement has a cartesian product between FROM:sqlalchemy.exc.SAWarning:airflow",
+    'error:Coercing Subquery object into a select\(\) for use in IN\(\):sqlalchemy.exc.SAWarning:airflow',
     "ignore::DeprecationWarning:flask_appbuilder.filemanager",
     "ignore::DeprecationWarning:flask_appbuilder.widgets",
     # FAB do not support SQLAclhemy 2


Test patch:
diff --git a/tests/providers/airbyte/hooks/test_airbyte.py b/tests/providers/airbyte/hooks/test_airbyte.py
index 18d935bda3..6cf211909e 100644
--- a/tests/providers/airbyte/hooks/test_airbyte.py
+++ b/tests/providers/airbyte/hooks/test_airbyte.py
@@ -67,6 +67,23 @@ class TestAirbyteHook:
         assert resp.status_code == 200
         assert resp.json() == self._mock_sync_conn_success_response_body
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "host, port, schema, expected_base_url, description",
+        [
+            ("test-airbyte", 8001, "http", "http://test-airbyte:8001", "uri_with_port_and_schema"),
+            ("test-airbyte", None, "https", "https://test-airbyte", "uri_with_schema"),
+            ("test-airbyte", None, None, "http://test-airbyte", "uri_without_port_and_schema"),
+        ],
+    )
+    async def test_get_base_url(self, host, port, schema, expected_base_url, description):
+        conn_id = f"test_conn_{description}"
+        conn = Connection(conn_id=conn_id, conn_type="airbyte", host=host, port=port, schema=schema)
+        hook = AirbyteHook(airbyte_conn_id=conn_id)
+        db.merge_conn(conn)
+        _, base_url = await hook.get_headers_tenants_from_connection()
+        assert base_url == expected_base_url
+
     def test_get_job_status(self, requests_mock):
         requests_mock.post(
             self.get_job_endpoint, status_code=200, json=self._mock_job_status_success_response_body