Merge pull request #183 from youzh00/feature/issue-139-reentrantlocks

Replace synchronized blocks with ReentrantLock in JDBC driver and server classes (Issue #139)

Author: rrobetti

ojp-jdbc-driver/src/main/java/org/openjproxy/grpc/client/MultinodeUrlParser.java

@@ -29,7 +29,6 @@ public class MultinodeUrlParser {
     // Cache of statement services keyed by server configuration
     private static final Map<String, StatementService> statementServiceCache = new ConcurrentHashMap<>();
 
-
     /**
      * Helper class to return the service, connection URL, server endpoints, and datasource names.
      */
@@ -57,7 +56,8 @@ public static class ServiceAndUrl {
      * @param url the JDBC URL (should be already cleaned of datasource names)
      * @param dataSourceNames optional list of datasource names corresponding to each endpoint
      */
-    public synchronized static ServiceAndUrl getOrCreateStatementService(String url, List<String> dataSourceNames) {
+    // No synchronization needed: computeIfAbsent on ConcurrentHashMap provides the required atomicity.
+    public static ServiceAndUrl getOrCreateStatementService(String url, List<String> dataSourceNames) {
         try {
             // Try to parse as multinode URL
             List<ServerEndpoint> endpoints = MultinodeUrlParser.parseServerEndpoints(url, dataSourceNames);
@@ -121,7 +121,7 @@ public synchronized static ServiceAndUrl getOrCreateStatementService(String url,
      * Gets or creates a StatementService implementation based on the URL.
      * Backward compatibility method without datasource names.
      */
-    public synchronized static ServiceAndUrl getOrCreateStatementService(String url) {
+    public static ServiceAndUrl getOrCreateStatementService(String url) {
         return getOrCreateStatementService(url, null);
     }
 

ojp-jdbc-driver/src/main/java/org/openjproxy/jdbc/ResultSet.java

@@ -39,7 +39,6 @@
 import java.util.Map;
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicInteger;
-
 import static org.openjproxy.grpc.client.GrpcExceptionHandler.handle;
 
 @Slf4j
@@ -723,7 +722,7 @@ public boolean isAfterLast() throws SQLException {
     }
 
     @Override
-    public synchronized boolean isFirst() throws SQLException {
+    public boolean isFirst() throws SQLException {
         log.debug("isFirst called");
         if (this.inProxyMode) {
             return super.isFirst();

ojp-jdbc-driver/src/main/java/org/openjproxy/jdbc/xa/OjpXAConnection.java

@@ -22,6 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Implementation of XAConnection that connects to the OJP server for XA operations.
@@ -47,6 +48,7 @@ public class OjpXAConnection implements XAConnection, ServerHealthListener {
     private List<String> serverEndpoints;
     private final List<ConnectionEventListener> listeners = new ArrayList<>();
     private String boundServerAddress; // Phase 2: Track which server this connection is bound to
+    private final ReentrantLock sessionLock = new ReentrantLock();
 
     public OjpXAConnection(StatementService statementService, String url, String user, String password, Properties properties, List<String> serverEndpoints) {
         log.debug("Creating OjpXAConnection for URL: {}", url);
@@ -73,12 +75,15 @@ public OjpXAConnection(StatementService statementService, String url, String use
      * Lazily create the server-side session when first needed.
      * This avoids creating sessions that may never be used.
      */
-    private synchronized SessionInfo getOrCreateSession() throws SQLException {
+    private SessionInfo getOrCreateSession() throws SQLException {
         if (sessionInfo != null) {
             return sessionInfo;
         }
-        
+        sessionLock.lock();
         try {
+            if (sessionInfo != null) {
+                return sessionInfo;
+            }
             // Connect to server with XA flag enabled
             ConnectionDetails.Builder connBuilder = ConnectionDetails.newBuilder()
                     .setUrl(url)
@@ -115,6 +120,8 @@ private synchronized SessionInfo getOrCreateSession() throws SQLException {
         } catch (Exception e) {
             log.error("Failed to create XA connection session", e);
             throw new SQLException("Failed to create XA connection session", e);
+        }finally {
+            sessionLock.unlock();
         }
     }
 
@@ -125,17 +132,22 @@ private synchronized SessionInfo getOrCreateSession() throws SQLException {
      * @return The new SessionInfo
      * @throws SQLException if session recreation fails
      */
-    synchronized SessionInfo recreateSession() throws SQLException {
-        log.info("Recreating XA session (previous session: {})", 
-                sessionInfo != null ? sessionInfo.getSessionUUID() : "none");
-        
-        // Clear existing session
-        sessionInfo = null;
-        boundServerAddress = null;
-        xaResource = null; // Force recreation of XAResource with new session
-        
-        // Create new session (will use round-robin to select a different server)
-        return getOrCreateSession();
+     SessionInfo recreateSession() throws SQLException {
+        sessionLock.lock();
+        try {
+            log.info("Recreating XA session (previous session: {})",
+                    sessionInfo != null ? sessionInfo.getSessionUUID() : "none");
+
+            // Clear existing session
+            sessionInfo = null;
+            boundServerAddress = null;
+            xaResource = null; // Force recreation of XAResource with new session
+
+            // Create new session (will use round-robin to select a different server)
+            return getOrCreateSession();
+        } finally {
+            sessionLock.unlock();
+        }
     }
 
     @Override

ojp-jdbc-driver/src/main/java/org/openjproxy/jdbc/xa/OjpXADataSource.java

@@ -17,6 +17,7 @@
 import java.sql.SQLFeatureNotSupportedException;
 import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Logger;
 
 /**
@@ -57,6 +58,8 @@ public class OjpXADataSource implements XADataSource {
     private String dataSourceName;
     private List<String> serverEndpoints;
 
+    private static final ReentrantLock initLock = new ReentrantLock();
+
     public OjpXADataSource() {
         log.debug("Creating OjpXADataSource");
     }
@@ -73,46 +76,56 @@ public OjpXADataSource(String url, String user, String password) {
      * This is done lazily when the first XA connection is requested.
      * The GRPC channel is opened once and reused by all XA connections from this datasource.
      */
-    private synchronized void initializeIfNeeded() throws SQLException {
+    private void initializeIfNeeded() throws SQLException {
+        // Fast path - no lock needed if already initialized
         if (statementService != null) {
             return; // Already initialized
         }
-        
-        if (url == null || url.isEmpty()) {
-            throw new SQLException("URL is not set");
-        }
-        
-        // Parse URL to extract datasource name and clean URL
-        UrlParser.UrlParseResult urlParseResult = UrlParser.parseUrlWithDataSource(url);
-        this.cleanUrl = urlParseResult.cleanUrl;
-        this.dataSourceName = urlParseResult.dataSourceName;
-        
-        log.debug("Parsed URL - clean: {}, dataSource: {}", cleanUrl, dataSourceName);
-
-        // Load ojp.properties file and extract datasource-specific configuration
-        Properties ojpProperties = DatasourcePropertiesLoader.loadOjpPropertiesForDataSource(dataSourceName);
-        if (ojpProperties != null && !ojpProperties.isEmpty()) {
-            // Merge ojp.properties with any manually set properties
-            for (String key : ojpProperties.stringPropertyNames()) {
-                if (!properties.containsKey(key)) {
-                    properties.setProperty(key, ojpProperties.getProperty(key));
+
+        initLock.lock();
+        try {
+            // Double-check inside lock
+            if (statementService != null) {
+                return;
+            }
+
+            if (url == null || url.isEmpty()) {
+                throw new SQLException("URL is not set");
+            }
+            // Parse URL to extract datasource name and clean URL
+            UrlParser.UrlParseResult urlParseResult = UrlParser.parseUrlWithDataSource(url);
+            this.cleanUrl = urlParseResult.cleanUrl;
+            this.dataSourceName = urlParseResult.dataSourceName;
+
+            log.debug("Parsed URL - clean: {}, dataSource: {}", cleanUrl, dataSourceName);
+
+            // Load ojp.properties file and extract datasource-specific configuration
+            Properties ojpProperties = DatasourcePropertiesLoader.loadOjpPropertiesForDataSource(dataSourceName);
+            if (ojpProperties != null && !ojpProperties.isEmpty()) {
+                // Merge ojp.properties with any manually set properties
+                for (String key : ojpProperties.stringPropertyNames()) {
+                    if (!properties.containsKey(key)) {
+                        properties.setProperty(key, ojpProperties.getProperty(key));
+                    }
                 }
+                log.debug("Loaded ojp.properties with {} properties for dataSource: {}", ojpProperties.size(), dataSourceName);
             }
-            log.debug("Loaded ojp.properties with {} properties for dataSource: {}", ojpProperties.size(), dataSourceName);
+
+            // Initialize StatementService - this will open the GRPC channel on first use
+            log.debug("Initializing StatementServiceGrpcClient for XA datasource: {}", dataSourceName);
+
+            // Detect multinode vs single-node configuration and get the URL to use for connection
+            MultinodeUrlParser.ServiceAndUrl serviceAndUrl = MultinodeUrlParser.getOrCreateStatementService(cleanUrl);
+            statementService = serviceAndUrl.getService();
+            this.serverEndpoints = serviceAndUrl.getServerEndpoints();
+
+            // The GRPC channel will be opened lazily on the first connect() call
+            // Since this StatementService instance is shared by all XA connections from this datasource,
+            // the channel is opened once and reused
+            log.info("StatementService initialized for datasource: {}. GRPC channel will open on first use.", dataSourceName);
+        } finally {
+            initLock.unlock();
         }
-        
-        // Initialize StatementService - this will open the GRPC channel on first use
-        log.debug("Initializing StatementServiceGrpcClient for XA datasource: {}", dataSourceName);
-
-        // Detect multinode vs single-node configuration and get the URL to use for connection
-        MultinodeUrlParser.ServiceAndUrl serviceAndUrl = MultinodeUrlParser.getOrCreateStatementService(cleanUrl);
-        statementService = serviceAndUrl.getService();
-        this.serverEndpoints = serviceAndUrl.getServerEndpoints();
-
-        // The GRPC channel will be opened lazily on the first connect() call
-        // Since this StatementService instance is shared by all XA connections from this datasource,
-        // the channel is opened once and reused
-        log.info("StatementService initialized for datasource: {}. GRPC channel will open on first use.", dataSourceName);
     }
 
 

ojp-jdbc-driver/src/test/java/openjproxy/jdbc/testutil/SQLServerTestContainer.java

@@ -2,6 +2,8 @@
 
 import org.testcontainers.containers.MSSQLServerContainer;
 
+import java.util.concurrent.locks.ReentrantLock;
+
 /**
  * Singleton SQL Server test container for all SQL Server integration tests.
  * This ensures that all tests share the same SQL Server instance to improve test performance
@@ -15,35 +17,46 @@ public class SQLServerTestContainer {
     private static MSSQLServerContainer<?> container;
     private static boolean isStarted = false;
     private static boolean shutdownHookRegistered = false;
-    
+
+    private static ReentrantLock initLock = new ReentrantLock();
     /**
      * Gets or creates the shared SQL Server test container instance.
      * The container is automatically started on first access.
      * 
      * @return the shared MSSQLServerContainer instance
      */
-    public static synchronized MSSQLServerContainer<?> getInstance() {
-        if (container == null) {
-            container = new MSSQLServerContainer<>(MSSQL_IMAGE)
-                    .acceptLicense();
+    public static  MSSQLServerContainer<?> getInstance() {
+        // Fast-path: if container already created and running, return it without locking
+        MSSQLServerContainer<?> local = container;
+        if (local != null && local.isRunning()) {
+            return local;
         }
-        
-        if (!isStarted) {
-            container.start();
-            isStarted = true;
-            
-            // Add shutdown hook to stop container when JVM exits
-            if (!shutdownHookRegistered) {
-                Runtime.getRuntime().addShutdownHook(new Thread(() -> {
-                    if (container != null && container.isRunning()) {
-                        container.stop();
-                    }
-                }));
-                shutdownHookRegistered = true;
+
+        initLock.lock();
+        try {
+            if (container == null) {
+                container = new MSSQLServerContainer<>(MSSQL_IMAGE).acceptLicense();
             }
+
+            if (!isStarted) {
+                container.start();
+                isStarted = true;
+
+                // Add shutdown hook to stop container when JVM exits
+                if (!shutdownHookRegistered) {
+                    Runtime.getRuntime().addShutdownHook(new Thread(() -> {
+                        if (container != null && container.isRunning()) {
+                            container.stop();
+                        }
+                    }));
+                    shutdownHookRegistered = true;
+                }
+            }
+
+            return container;
+        }finally {
+            initLock.unlock();
         }
-        
-        return container;
     }
 
     /**

ojp-jdbc-driver/src/test/java/openjproxy/jdbc/testutil/TestDBUtils.java

@@ -11,6 +11,7 @@
 import java.sql.ResultSetMetaData;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -165,13 +166,13 @@ public void close() {
      * Simple Xid implementation for XA transactions in tests.
      */
     private static class SimpleXid implements Xid {
-        private static int counter = 0;
+        // AtomicInteger replaces synchronized(SimpleXid.class) for thread-safe ID generation
+        // and avoids virtual-thread pinning
+        private static AtomicInteger counter = new AtomicInteger(0);;
         private final int id;
 
         public SimpleXid() {
-            synchronized (SimpleXid.class) {
-                this.id = counter++;
-            }
+            this.id = counter.getAndIncrement();
         }
 
         @Override

ojp-jdbc-driver/src/test/java/org/openjproxy/grpc/client/MultinodeXAIntegrationTest.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
 
 import static org.junit.jupiter.api.Assumptions.assumeFalse;
 
@@ -47,6 +48,7 @@ public class MultinodeXAIntegrationTest {
     private static AtomicInteger totalFailedQueries = new AtomicInteger(0);
     private static AtomicInteger nonConnectivityFailedQueries = new AtomicInteger(0);
     private static ExecutorService queryExecutor = new RoundRobinExecutorService(100);
+    private static final ReentrantLock multinodeXaLock = new ReentrantLock();
 
 
     private static AtomikosDataSourceBean xaDataSource;
@@ -252,36 +254,37 @@ protected static Connection getConnection(String driverClass, String url, String
         // Load driver if needed
         Class.forName(driverClass);
         Connection conn;
-        synchronized (MultinodeIntegrationTest.class) {
+        multinodeXaLock.lock();
+        try {
             if (xaDataSource == null) {
                 OjpXADataSource xaDataSourceImpl = new OjpXADataSource();
                 //Kept for reference testing with Postgres native XA
                 //PGXADataSource xaDataSourceImpl = new PGXADataSource();
                 xaDataSourceImpl.setUrl(url);
                 xaDataSourceImpl.setUser(user);
                 xaDataSourceImpl.setPassword(password);
-
                 xaDataSource = new AtomikosDataSourceBean();
                 xaDataSource.setUniqueResourceName("ATOMIKOS_OJP_XA_DS");
                 xaDataSource.setXaDataSource(xaDataSourceImpl);
                 xaDataSource.setMinPoolSize(ATOMIKOS_MIN_POOL_SIZE);
                 xaDataSource.setMaxPoolSize(ATOMIKOS_MAX_POOL_SIZE);
                 log.info("✓ Atomikos XA DataSource initialized");
             }
-
             // Block below is to increase the chances getting different connections to better test multinode balancing
             // as per Atomikos tends to reuse the very same connection whenever possible.
-            int num = (int)(Math.random() * 10) + 1;
+            int num = (int) (Math.random() * 10) + 1;
             List<Connection> connectionList = new ArrayList<>();
             for (int i = 0; i < num; i++) {
                 connectionList.add(xaDataSource.getConnection());
             }
-            conn = connectionList.remove(connectionList.size()-1);
-            for(Connection c : connectionList){
+            conn = connectionList.remove(connectionList.size() - 1);
+            for (Connection c : connectionList) {
                 c.close();
             }
+            return conn;
+        } finally {
+            multinodeXaLock.unlock();
         }
-        return conn;
     }
 
     private static void runExactQuerySequence(int threadNum, String driverClass, String url, String user, String password) {