diff --git a/providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py b/providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py
index 56449fec0bc85..550fc7df975fc 100644
--- a/providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py
+++ b/providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py
@@ -33,14 +33,14 @@
 from airflow.providers.google.cloud.hooks.dataproc import DataprocAsyncHook, DataprocHook
 from airflow.providers.google.cloud.utils.dataproc import DataprocOperationType
 from airflow.providers.google.common.hooks.base_google import PROVIDE_PROJECT_ID
-from airflow.providers.google.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_3_0_PLUS
+from airflow.providers.google.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_3_PLUS
 from airflow.triggers.base import BaseTrigger, TriggerEvent
 from airflow.utils.state import TaskInstanceState
 
 if TYPE_CHECKING:
     from sqlalchemy.orm.session import Session
 
-if not AIRFLOW_V_3_3_0_PLUS:
+if not AIRFLOW_V_3_3_PLUS:
     from sqlalchemy import select
 
     from airflow.models.taskinstance import TaskInstance
@@ -132,7 +132,7 @@ async def on_kill(self) -> None:
             )
             self.log.info("Job: %s is cancelled.", self.job_id)
 
-    if not AIRFLOW_V_3_3_0_PLUS:
+    if not AIRFLOW_V_3_3_PLUS:
 
         @provide_session
         def get_task_instance(self, session: Session) -> TaskInstance:
@@ -212,7 +212,7 @@ async def run(self):
                 {"job_id": self.job_id, "job_state": JobStatus.State(state).name, "job": Job.to_dict(job)}
             )
         except asyncio.CancelledError:
-            if not AIRFLOW_V_3_3_0_PLUS:
+            if not AIRFLOW_V_3_3_PLUS:
                 self.log.info("Task got cancelled.")
                 try:
                     if self.job_id and self.cancel_on_kill and await self.safe_to_cancel():
@@ -284,7 +284,7 @@ async def on_kill(self) -> None:
             )
             self.log.info("Job: %s is cancelled.", self.job_id)
 
-    if not AIRFLOW_V_3_3_0_PLUS:
+    if not AIRFLOW_V_3_3_PLUS:
 
         @provide_session
         def get_task_instance(self, session: Session) -> TaskInstance:
@@ -371,7 +371,7 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
                 {"job_id": self.job_id, "job_state": JobStatus.State(state).name, "job": Job.to_dict(job)}
             )
         except asyncio.CancelledError:
-            if not AIRFLOW_V_3_3_0_PLUS:
+            if not AIRFLOW_V_3_3_PLUS:
                 self.log.info("Task got cancelled.")
                 try:
                     if self.job_id and self.cancel_on_kill and await self.safe_to_cancel():
diff --git a/providers/google/src/airflow/providers/google/version_compat.py b/providers/google/src/airflow/providers/google/version_compat.py
index 47dbdc0aa745e..57d8840c9efb2 100644
--- a/providers/google/src/airflow/providers/google/version_compat.py
+++ b/providers/google/src/airflow/providers/google/version_compat.py
@@ -35,7 +35,7 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
 AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 0, 0)
 AIRFLOW_V_3_1_PLUS = get_base_airflow_version_tuple() >= (3, 1, 0)
 # BaseTrigger.on_kill() was introduced in 3.3.0; use this flag to guard trigger migrations
-AIRFLOW_V_3_3_0_PLUS = get_base_airflow_version_tuple() >= (3, 3, 0)
+AIRFLOW_V_3_3_PLUS = get_base_airflow_version_tuple() >= (3, 3, 0)
 
 # Version-compatible imports
 # BaseOperator: Use 3.1+ due to xcom_push method missing in SDK BaseOperator 3.0.x
@@ -49,4 +49,4 @@ def get_base_airflow_version_tuple() -> tuple[int, int, int]:
 
 
 # Explicitly export these imports to protect them from being removed by linters
-__all__ = ["AIRFLOW_V_3_0_PLUS", "AIRFLOW_V_3_1_PLUS", "AIRFLOW_V_3_3_0_PLUS", "BaseOperator"]
+__all__ = ["AIRFLOW_V_3_0_PLUS", "AIRFLOW_V_3_1_PLUS", "AIRFLOW_V_3_3_PLUS", "BaseOperator"]