FIX: Only pass connection to sqlalchemy engine in JdbcHook

airflow/providers/common/sql/hooks/sql.py

@@ -272,10 +272,6 @@ def get_sqlalchemy_engine(self, engine_kwargs=None):
         :param engine_kwargs: Kwargs used in :func:`~sqlalchemy.create_engine`.
         :return: the created engine.
         """
-        if engine_kwargs is None:
-            engine_kwargs = {}
-        engine_kwargs["creator"] = self.get_conn
-
         try:
             url = self.sqlalchemy_url
         except NotImplementedError:

airflow/providers/jdbc/hooks/jdbc.py

@@ -163,6 +163,19 @@ def sqlalchemy_url(self) -> URL:
             database=conn.schema,
         )
 
+    def get_sqlalchemy_engine(self, engine_kwargs=None):
+        """
+        Get an sqlalchemy_engine object.
+
+        :param engine_kwargs: Kwargs used in :func:`~sqlalchemy.create_engine`.
+        :return: the created engine.
+        """
+        if engine_kwargs is None:
+            engine_kwargs = {}
+        engine_kwargs["creator"] = self.get_conn
+
+        return super().get_sqlalchemy_engine(**engine_kwargs)
+
     def get_conn(self) -> jaydebeapi.Connection:
         conn: Connection = self.get_connection(self.get_conn_id())
         host: str = conn.host