Fix cursor leaks and verify the fix when testing

Author: Michael-A-McMahon

src/main/java/oracle/r2dbc/impl/OracleReactiveJdbcAdapter.java

@@ -511,7 +511,10 @@ private static void configureExtendedOptions(
 
     // Apply any extended options as connection properties
     for (Option<CharSequence> option : SUPPORTED_CONNECTION_PROPERTY_OPTIONS) {
-      CharSequence value = options.getValue(option);
+      // Using Object as the value type allows options to be set as types like
+      // Boolean or Integer. These types make sense for numeric or boolean
+      // connection property values, such as statement cache size, or enable x.
+      Object value = options.getValue(option);
       if (value != null) {
         runOrHandleSQLException(() ->
           oracleDataSource.setConnectionProperty(

src/main/java/oracle/r2dbc/impl/OracleResultImpl.java

@@ -35,6 +35,7 @@
 
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.getOrHandleSQLException;
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.requireNonNull;
+import static oracle.r2dbc.impl.OracleR2dbcExceptions.runOrHandleSQLException;
 
 /**
  * <p>
@@ -90,9 +91,13 @@ <T> Publisher<T> publishRows(
   }
 
   /**
+   * <p>
    * Creates a {@code Result} that either publishes a {@code ResultSet} of
    * row data from a query, or publishes an update count as an empty stream.
-   *
+   * </p><p>
+   * The {@link java.sql.Statement} that created the {@code resultSet} is closed
+   * when the returned result is fully consumed.
+   * </p>
    * @param adapter Adapts {@code ResultSet} API calls into reactive streams.
    *   Not null.
    * @param resultSet Row data to publish
@@ -105,19 +110,29 @@ public static Result createQueryResult(
 
       @Override
       Publisher<Integer> publishUpdateCount() {
+        runOrHandleSQLException(() ->
+          resultSet.getStatement().close());
         return Mono.empty();
       }
 
       @Override
       <T> Publisher<T> publishRows(
         BiFunction<Row, RowMetadata, ? extends T> mappingFunction) {
 
+        // Obtain a reference to the statement before the ResultSet is
+        // logically closed by its row publisher. The statement is closed when
+        // the publisher terminates.
+        java.sql.Statement jdbcStatement =
+          getOrHandleSQLException(resultSet::getStatement);
+
         OracleRowMetadataImpl metadata = new OracleRowMetadataImpl(
           getOrHandleSQLException(resultSet::getMetaData));
 
-        return Flux.from(adapter.publishRows(resultSet, jdbcRow ->
+        return Flux.<T>from(adapter.publishRows(resultSet, jdbcRow ->
           mappingFunction.apply(
-            new OracleRowImpl(jdbcRow, metadata, adapter), metadata)));
+            new OracleRowImpl(jdbcRow, metadata, adapter), metadata)))
+          .doFinally(signalType ->
+            runOrHandleSQLException(jdbcStatement::close));
       }
     };
   }
@@ -128,10 +143,15 @@ <T> Publisher<T> publishRows(
    * {@link PreparedStatement#getGeneratedKeys()} {@code ResultSet}, or
    * publishes an {@code updateCount}.
    * </p><p>
+   * The {@link java.sql.Statement} that created the {@code ResultSet} is closed
+   * when the {@code Publisher} returned by this method emits a
+   * {@code Result}.
+   * </p><p>
    * For compliance with R2DBC standards, a {@code Row} of generated column
    * values will remain valid after the {@code Connection} that created them
    * is closed. This behavior is verified by version 0.8.2 of
-   * {@code io.r2dbc.spi.test.TestKit#returnGeneratedValues()}.
+   * {@code io.r2dbc.spi.test.TestKit#returnGeneratedValues()}. The {@code Rows}
+   * of generated value
    * </p>
    *
    * @implNote
@@ -167,16 +187,25 @@ public static Publisher<Result> createGeneratedValuesResult(
 
     // Avoid invoking ResultSet.getMetaData() on an empty ResultSet, it may
     // throw a SQLException
-    if (! getOrHandleSQLException(values::isBeforeFirst))
+    if (! getOrHandleSQLException(values::isBeforeFirst)) {
+      runOrHandleSQLException(() -> values.getStatement().close());
       return Mono.just(createUpdateCountResult(updateCount));
+    }
 
     // Obtain metadata before the ResultSet is closed by publishRows(...)
     OracleRowMetadataImpl metadata =
       new OracleRowMetadataImpl(getOrHandleSQLException(values::getMetaData));
 
+    // Obtain a reference to the statement before the ResultSet is
+    // logically closed by its row publisher. The statement is closed when
+    // the publisher terminates.
+    java.sql.Statement jdbcStatement =
+      getOrHandleSQLException(values::getStatement);
+
     return Flux.from(adapter.publishRows(
       values, ReactiveJdbcAdapter.JdbcRow::copy))
       .collectList()
+      .doFinally(signalType -> runOrHandleSQLException(jdbcStatement::close))
       .map(cachedRows -> new OracleResultImpl() {
 
         @Override

src/main/java/oracle/r2dbc/impl/OracleStatementImpl.java

@@ -41,6 +41,7 @@
 import java.util.Queue;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
+import java.util.function.Function;
 
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.getOrHandleSQLException;
 import static oracle.r2dbc.impl.OracleR2dbcExceptions.requireNonNull;
@@ -546,8 +547,8 @@ private Publisher<Result> createResultPublisher(
 
   /**
    * <p>
-   * Executes a JDBC statement with a single, non-batched, set of parameters.
-   * If the execution returns a {@link java.sql.ResultSet} then the
+   * Executes a JDBC statement with a single, non-batched, set of {@code
+   * bindValues}. If the execution results in a {@link java.sql.ResultSet} then the
    * {@code jdbcStatement} is closed after the {@code ResultSet} is fully
    * consumed by {@link Result#map(BiFunction)}. Otherwise, if the execution
    * only produces an update count, then the {@code jdbcStatement} is closed
@@ -569,7 +570,6 @@ private Publisher<Result> executeSingle(
     return Mono.from(adapter.publishSQLExecution(jdbcStatement))
       .map(isResultSet -> {
         if (isResultSet) {
-          runOrHandleSQLException(jdbcStatement::closeOnCompletion);
           return OracleResultImpl.createQueryResult(
               adapter, getOrHandleSQLException(jdbcStatement::getResultSet));
         }
@@ -614,22 +614,27 @@ private Publisher<Result> executeBatch(
 
     // Execute the batch. The execution won't return a ResultSet, so the JDBC
     // statement is closed immediately after the execution completes.
-    runOrHandleSQLException(jdbcStatement::closeOnCompletion);
     return Flux.from(adapter.publishBatchUpdate(jdbcStatement))
       .map(updateCount ->
-        OracleResultImpl.createUpdateCountResult(Math.toIntExact(updateCount)));
+        OracleResultImpl.createUpdateCountResult(Math.toIntExact(updateCount)))
+      .doFinally(signalType -> runOrHandleSQLException(jdbcStatement::close));
   }
 
   /**
    * <p>
-   * Executes a key generating {@code jdbcStatement} for each set
-   * of bind values in a {@code batch}. The {@code jdbcStatement} is closed
-   * after all executions have completed. If any execution results in an
-   * error, subsequent executions are skipped.
+   * Executes a JDBC statement with a single, non-batched, set of {@code
+   * bindValues}. If the execution results in a {@link java.sql.ResultSet} then
+   * the {@code jdbcStatement} is closed after the {@code ResultSet} is fully
+   * consumed by {@link Result#map(BiFunction)}. If the execution results in
+   * an update count and/or values generated by DML, then the {@code
+   * jdbcStatement} is closed after the returned {@code Publisher} emits a
+   * {@code Result} with all generated values cached, such that
+   * {@link Result#map(BiFunction)} may be called after the database connection
+   * has been closed.
    * </p><p>
-   * The returned {@code Publisher} emits a {@code Result} with an update
-   * count and generated values for each execution of the {@code
-   * jdbcStatement}.
+   * The returned publisher initiates SQL execution <i>the first time</i> a
+   * subscriber subscribes, before the subscriber emits a {@code request}
+   * signal.
    * </p>
    *
    * @implNote The 21c Oracle JDBC Driver does not support batch DML when
@@ -638,25 +643,23 @@ private Publisher<Result> executeBatch(
    *
    * @param jdbcStatement A JDBC statement
    * @param bindValues A set of bind values
-   * @return A publisher that emits the {@code Results} of executing the
-   * JDBC statement for each set of bind values in the {@code batch}
+   * @return A publisher that emits the {@code Result} of executing the
+   * JDBC statement.
    */
   private Publisher<Result> executeGeneratingValues(
     PreparedStatement jdbcStatement, Object[] bindValues) {
 
     setJdbcBindValues(bindValues, jdbcStatement);
     return Mono.from(adapter.publishSQLExecution(jdbcStatement))
-      .flatMap(isResultSet -> {
-        runOrHandleSQLException(jdbcStatement::closeOnCompletion);
-        return isResultSet
+      .flatMap(isResultSet ->
+        isResultSet
           ? Mono.just(OracleResultImpl.createQueryResult(
               adapter,
               getOrHandleSQLException(jdbcStatement::getResultSet)))
           : Mono.from(OracleResultImpl.createGeneratedValuesResult(
               adapter,
               getOrHandleSQLException(jdbcStatement::getUpdateCount),
-              getOrHandleSQLException(jdbcStatement::getGeneratedKeys)));
-      });
+              getOrHandleSQLException(jdbcStatement::getGeneratedKeys))));
   }
 
   /**

src/test/java/oracle/r2dbc/DatabaseConfig.java

@@ -25,6 +25,8 @@
 import io.r2dbc.spi.ConnectionFactories;
 import io.r2dbc.spi.ConnectionFactory;
 import io.r2dbc.spi.ConnectionFactoryOptions;
+import io.r2dbc.spi.Option;
+import oracle.jdbc.OracleConnection;
 import oracle.r2dbc.util.SharedConnectionFactory;
 import org.reactivestreams.Publisher;
 
@@ -216,6 +218,11 @@ public static int databaseVersion() {
           .option(ConnectionFactoryOptions.DATABASE, SERVICE_NAME)
           .option(ConnectionFactoryOptions.USER, USER)
           .option(ConnectionFactoryOptions.PASSWORD, PASSWORD)
+          // Disable statement caching in order to verify cursor closing;
+          // Cached statements don't close their cursors
+          .option(Option.valueOf(
+            OracleConnection.CONNECTION_PROPERTY_IMPLICIT_STATEMENT_CACHE_SIZE),
+            0)
           .build());
 
       SHARED_CONNECTION_FACTORY = new SharedConnectionFactory(

src/test/java/oracle/r2dbc/impl/OracleConnectionImplTest.java

@@ -559,7 +559,7 @@ public void testSetAutoCommit() {
             selectInSessionB);
         }
         finally {
-          awaitNone(sessionA.close());
+          awaitNone(sessionB.close());
         }
       }
       finally {