Made get_conn in JdbcHook threadsafe to avoid OSError: JVM is already started

providers/src/airflow/providers/jdbc/hooks/jdbc.py

@@ -28,6 +28,7 @@
 
 from airflow.exceptions import AirflowException
 from airflow.providers.common.sql.hooks.sql import DbApiHook
+from wrapt import synchronized
 
 if TYPE_CHECKING:
     from airflow.models.connection import Connection
@@ -177,6 +178,7 @@ def get_sqlalchemy_engine(self, engine_kwargs=None):
 
         return super().get_sqlalchemy_engine(engine_kwargs)
 
+    @synchronized
     def get_conn(self) -> jaydebeapi.Connection:
         conn: Connection = self.connection
         host: str = conn.host

providers/tests/jdbc/hooks/test_jdbc.py

@@ -20,12 +20,14 @@
 import json
 import logging
 import sqlite3
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from threading import current_thread
+from time import sleep
 from unittest import mock
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, patch, MagicMock
 
 import jaydebeapi
 import pytest
-
 from airflow.exceptions import AirflowException
 from airflow.models import Connection
 from airflow.providers.jdbc.hooks.jdbc import JdbcHook, suppress_and_warn
@@ -54,10 +56,17 @@ def get_hook(
             **conn_params,
         }
     )
+    jvm_started = False
 
     class MockedJdbcHook(JdbcHook):
         @classmethod
         def get_connection(cls, conn_id: str) -> Connection:
+            # nonlocal jvm_started
+            #
+            # if jvm_started:
+            #     raise OSError("JVM already started")
+            #
+            # jvm_started = True
             return connection
 
     hook = MockedJdbcHook(**hook_params)
@@ -229,3 +238,39 @@ def test_get_sqlalchemy_engine_verify_creator_is_being_used(self):
             jdbc_hook.get_conn = lambda: connection
             engine = jdbc_hook.get_sqlalchemy_engine()
             assert engine.connect().connection.connection == connection
+
+    def test_get_conn_thread_safety(self):
+        mock_conn = MagicMock()
+        open_connections = 0
+
+        def connect_side_effect(*args, **kwargs):
+            nonlocal open_connections
+            open_connections += 1
+            logging.debug("Thread %s has %s open connections", current_thread().name, open_connections)
+
+            try:
+                if open_connections > 1:
+                    raise OSError("JVM is already started")
+            finally:
+                sleep(0.1)  # wait a bit before releasing the connection again
+                open_connections -= 1
+
+            return mock_conn
+
+        with patch.object(jaydebeapi, "connect", side_effect=connect_side_effect) as mock_connect:
+            jdbc_hook = get_hook()
+
+            def call_get_conn():
+                conn = jdbc_hook.get_conn()
+                assert conn is mock_conn
+
+            with ThreadPoolExecutor(max_workers=10) as executor:
+                futures = []
+
+                for _ in range(0, 10):
+                    futures.append(executor.submit(call_get_conn))
+
+                for future in as_completed(futures):
+                    future.result()  # This will raise OSError if get_conn isn't threadsafe
+
+            assert mock_connect.call_count == 10