fix[PDI-20410]: Refactor Database class to replace ReentrantLock with synchronized block for connection handling (#10073)

* fix[PDI-20410]: Refactor Database class to replace ReentrantLock with synchronized block for connection handling

* test: fix failing unit test

Author: befc

core/src/main/java/org/pentaho/di/core/database/Database.java

@@ -44,8 +44,6 @@
 import java.util.Map;
 import java.util.Properties;
 import java.util.Set;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
 
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.vfs2.FileObject;
@@ -181,8 +179,6 @@ public class Database implements VariableSpace, LoggingObjectInterface, Closeabl
   private DataSource dataSource;
   private String ownerName;
 
-  private static final Lock lock = new ReentrantLock();
-
   static {
     initValueMetaPluginClasses();
     // listen for changes to make sure the valueMetaPluginClasses list is accurate.
@@ -2732,26 +2728,27 @@ public Object[] getRow( ResultSet rs, ResultSetMetaData dummy, RowMetaInterface
     long startTime = System.currentTimeMillis();
 
     try {
-      lock.lock();
-
-      int nrcols = rowInfo.size();
-      Object[] data = RowDataUtil.allocateRowData( nrcols );
+      // PDI-19750 - synchronize on the connection to prevent data inconsistency
+      // when multiple Database objects share the same connection and access ResultSets concurrently
+      synchronized ( connection ) {
+        int nrcols = rowInfo.size();
+        Object[] data = RowDataUtil.allocateRowData( nrcols );
 
-      if ( rs.next() ) {
-        for ( int i = 0; i < nrcols; i++ ) {
-          ValueMetaInterface val = rowInfo.getValueMeta( i );
+        if ( rs.next() ) {
+          for ( int i = 0; i < nrcols; i++ ) {
+            ValueMetaInterface val = rowInfo.getValueMeta( i );
 
-          data[ i ] = databaseMeta.getValueFromResultSet( rs, val, i );
+            data[ i ] = databaseMeta.getValueFromResultSet( rs, val, i );
+          }
+        } else {
+          data = null;
         }
-      } else {
-        data = null;
-      }
 
-      return data;
+        return data;
+      }
     } catch ( Exception ex ) {
       throw new KettleDatabaseException( "Couldn't get row from result set", ex );
     } finally {
-      lock.unlock();
       if ( log.isGatheringMetrics() ) {
         long time = System.currentTimeMillis() - startTime;
         log.snap( Metrics.METRIC_DATABASE_GET_ROW_SUM_TIME, databaseMeta.getName(), time );

core/src/test/java/org/pentaho/di/core/database/DatabaseTest.java

@@ -270,9 +270,11 @@ public void testGetLookupCallPSpassed() throws SQLException, KettleDatabaseExcep
     when( rsMetaData.getColumnCount() ).thenReturn( 0 );
     when( ps.getMetaData() ).thenReturn( rsMetaData );
 
-    Database db = new Database( log, meta );
-    db.getLookup( ps );
-    verify( rsMetaData, times( 1 ) ).getColumnCount();
+    try ( Database db = new Database( log, meta ) ) {
+      db.setConnection( conn );
+      db.getLookup( ps );
+      verify( rsMetaData, times( 1 ) ).getColumnCount();
+    }
   }
 
   @Test