Update to ojdbc 23.5

Author: Michael-A-McMahon

pom.xml

@@ -65,7 +65,7 @@
 
   <properties>
     <java.version>11</java.version>
-    <ojdbc.version>23.4.0.24.05</ojdbc.version>
+    <ojdbc.version>23.5.0.24.07</ojdbc.version>
     <r2dbc.version>1.0.0.RELEASE</r2dbc.version>
     <reactor.version>3.5.11</reactor.version>
     <reactive-streams.version>1.0.3</reactive-streams.version>

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

@@ -56,6 +56,9 @@
  */
 final class OracleBatchImpl implements Batch {
 
+  /** The OracleConnectionImpl that created this Batch */
+  private final OracleConnectionImpl r2dbcConnection;
+
   /** Adapts Oracle JDBC Driver APIs into Reactive Streams APIs */
   private final ReactiveJdbcAdapter adapter;
 
@@ -83,12 +86,11 @@ final class OracleBatchImpl implements Batch {
    * @param jdbcConnection JDBC connection to an Oracle Database. Not null.
    * @param adapter Adapts JDBC calls into reactive streams. Not null.
    */
-  OracleBatchImpl(
-    Duration timeout, Connection jdbcConnection, ReactiveJdbcAdapter adapter) {
+  OracleBatchImpl(Duration timeout, OracleConnectionImpl r2dbcConnection) {
     this.timeout = timeout;
-    this.jdbcConnection =
-      requireNonNull(jdbcConnection, "jdbcConnection is null");
-    this.adapter = requireNonNull(adapter, "adapter is null");
+    this.r2dbcConnection = r2dbcConnection;
+    this.jdbcConnection = r2dbcConnection.jdbcConnection();
+    this.adapter = r2dbcConnection.adapter();
   }
 
   /**
@@ -103,7 +105,7 @@ public Batch add(String sql) {
     requireOpenConnection(jdbcConnection);
     requireNonNull(sql, "sql is null");
     statements.add(
-      new OracleStatementImpl(sql, timeout, jdbcConnection, adapter));
+      new OracleStatementImpl(sql, timeout, r2dbcConnection));
     return this;
   }
 

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

@@ -38,6 +38,8 @@
 import java.sql.SQLException;
 import java.sql.Savepoint;
 import java.time.Duration;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 
 import static io.r2dbc.spi.IsolationLevel.READ_COMMITTED;
 import static io.r2dbc.spi.IsolationLevel.SERIALIZABLE;
@@ -126,6 +128,12 @@ final class OracleConnectionImpl implements Connection, Lifecycle {
    */
   private TransactionDefinition currentTransaction = null;
 
+  /**
+   * A queue of tasks that must complete before the {@link #jdbcConnection} is
+   * closed.
+   */
+  private final Queue<Publisher<?>> closeTasks = new ConcurrentLinkedQueue<>();
+
   /**
    * Constructs a new connection that uses the specified {@code adapter} to
    * perform database operations with the specified {@code jdbcConnection}.
@@ -369,7 +377,46 @@ else if (isReadOnly == null && name == null) {
    */
   @Override
   public Publisher<Void> close() {
-    return adapter.publishClose(jdbcConnection);
+
+    Publisher<Void> closeTasksPublisher = Mono.defer(() -> {
+      Publisher<?>[] closeTasksArray = closeTasks.toArray(Publisher<?>[]::new);
+      closeTasks.clear();
+
+      return Flux.concatDelayError(closeTasksArray).then();
+    });
+
+    return Flux.concatDelayError(
+      closeTasksPublisher,
+      adapter.publishClose(jdbcConnection));
+  }
+
+  /**
+   * <p>
+   * Adds a publisher that must be subscribed to and must terminate before
+   * closing the JDBC connection. This can be used to ensure that a publisher
+   * has completed a task before the {@link #jdbcConnection()} has been closed
+   * and becomes unusable.
+   * </p><p>
+   * <i> A call to this method should always be accompanied with a call to
+   * {@link #removeCloseTask(Publisher)}</i>: If the publisher is subscribed to
+   * and it terminates before {@link #close()} is called, then any reference
+   * to the Publisher must be removed so that it can be garbage collected.
+   * </p>
+   * @param publisher Publisher that must terminate before closing the JDBC
+   * connection. Not null.
+   */
+  void addCloseTask(Publisher<?> publisher) {
+    closeTasks.add(publisher);
+  }
+
+  /**
+   * Removes a publisher that was previously added with
+   * {@link #addCloseTask(Publisher)}.
+   *
+   * @param publisher Publisher to remove. Not null.
+   */
+  void removeCloseTask(Publisher<?> publisher) {
+    closeTasks.remove(publisher);
   }
 
   /**
@@ -417,7 +464,7 @@ public Publisher<Void> commitTransaction() {
   @Override
   public Batch createBatch() {
     requireOpenConnection(jdbcConnection);
-    return new OracleBatchImpl(statementTimeout, jdbcConnection, adapter);
+    return new OracleBatchImpl(statementTimeout, this);
   }
 
   /**
@@ -441,8 +488,7 @@ public Batch createBatch() {
   public Statement createStatement(String sql) {
     requireNonNull(sql, "sql is null");
     requireOpenConnection(jdbcConnection);
-    return new OracleStatementImpl(
-      sql, statementTimeout, jdbcConnection, adapter);
+    return new OracleStatementImpl(sql, statementTimeout, this);
   }
 
   /**
@@ -826,4 +872,24 @@ public Publisher<Void> preRelease() {
     });
   }
 
+  /**
+   * Returns the JDBC connection that this R2DBC connection executes database
+   * calls with.
+   *
+   * @return The JDBC connection that backs this R2DBC connection. Not null.
+   */
+  java.sql.Connection jdbcConnection() {
+    return jdbcConnection;
+  }
+
+  /**
+   * Returns the adapter that adapts the asynchronous API of the
+   * {@link #jdbcConnection()} that backs this R2DBC connection.
+   *
+   * @return The JDBC connection that backs this R2DBC connection. Not null.
+   */
+  ReactiveJdbcAdapter adapter() {
+    return adapter;
+  }
+
 }

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

@@ -197,6 +197,11 @@ final class OracleStatementImpl implements Statement {
   /** Adapts Oracle JDBC Driver APIs into Reactive Streams APIs */
   private final ReactiveJdbcAdapter adapter;
 
+  /**
+   * The instance of OracleConnectionImpl that created this statement.
+   */
+  private final OracleConnectionImpl r2dbcConnection;
+
   /**
    * SQL Language command that this statement executes. The command is
    * provided by user code and may include parameter markers.
@@ -255,15 +260,15 @@ final class OracleStatementImpl implements Statement {
    * @param sql SQL Language statement that may include parameter markers.
    * @param timeout Timeout applied to the execution of the constructed
    * {@code Statement}. Not null. Not negative.
-   * @param jdbcConnection JDBC connection to an Oracle Database.
-   * @param adapter Adapts JDBC calls into reactive streams.
+   * @param r2dbcConnection The OracleConnectionImpl that is creating this
+   * statement. Not null.
    */
   OracleStatementImpl(
-    String sql, Duration timeout, Connection jdbcConnection,
-    ReactiveJdbcAdapter adapter) {
+    String sql, Duration timeout, OracleConnectionImpl r2dbcConnection) {
     this.sql = sql;
-    this.jdbcConnection = jdbcConnection;
-    this.adapter = adapter;
+    this.r2dbcConnection = r2dbcConnection;
+    this.jdbcConnection = r2dbcConnection.jdbcConnection();
+    this.adapter = r2dbcConnection.adapter();
 
     // The SQL string is parsed to identify parameter markers and allocate the
     // bindValues array accordingly
@@ -987,13 +992,29 @@ private JdbcStatement(PreparedStatement preparedStatement, Object[] binds) {
       this.preparedStatement = preparedStatement;
       this.binds = binds;
 
+      // Work around for Oracle JDBC bug #37160069: The JDBC statement must be
+      // closed before closeAsyncOracle is called. This bug should be fixed
+      // by the 23.7 release of Oracle JDBC. The fix can be verified by the
+      // OracleConnectionImplTest.testSetStatementTimeout method (test won't
+      // fail, but look for an error in stderr). Typically, statement closing
+      // is a no-op if the connection is closed. However, if the statement
+      // executes SELECT ... FOR UPDATE, then JDBC will implicitly execute a
+      // commit() when the Statement (or really the ResultSet) is closed. This
+      // commit operation fails if the JDBC connection is already closed.
+      Publisher<Void> closePublisher = closeStatement();
+      r2dbcConnection.addCloseTask(closePublisher);
+
+      dependentCounter = new DependentCounter(Publishers.concatTerminal(
+        closePublisher,
+        Mono.fromRunnable(() ->
+          r2dbcConnection.removeCloseTask(closePublisher))));
+
       // Add this statement as a "party" (think j.u.c.Phaser) to the dependent
       // results by calling increment(). After the Result publisher returned by
       // execute() terminates, this statement "arrives" by calling decrement().
       // Calling decrement() after the Result publisher terminates ensures that
       // the JDBC statement can not be closed until all results have had a
       // chance to be emitted to user code.
-      dependentCounter = new DependentCounter(closeStatement());
       dependentCounter.increment();
     }
 
@@ -1864,7 +1885,24 @@ private Publisher<java.sql.Blob> convertBlobBind(
           Mono.from(adapter.publishBlobWrite(r2dbcBlob.stream(), jdbcBlob))
             .thenReturn(jdbcBlob),
         jdbcBlob -> {
-          addDeallocation(adapter.publishBlobFree(jdbcBlob));
+          Publisher<Void> freePublisher = adapter.publishBlobFree(jdbcBlob);
+
+          // Work around for Oracle JDBC bug #37160069: All LOBs need to be
+          // freed before closeAsyncOracle is called. This bug should be fixed
+          // by the 23.7 release of Oracle JDBC. The fix can be verified by the
+          // clobInsert and blobInsert methods in the TestKit class of the R2DBC
+          // SPI test: These tests will subscribe to Connection.close() before
+          // this freePublisher is subscribed to.
+          r2dbcConnection.addCloseTask(freePublisher);
+
+          addDeallocation(
+            Publishers.concatTerminal(
+              freePublisher,
+              Mono.fromRunnable(() ->
+                r2dbcConnection.removeCloseTask(freePublisher))));
+
+          // TODO: Why is discard() called here? It should be called by the
+          //  user who allocated the Blob, not by Oracle R2DBC.
           return r2dbcBlob.discard();
         });
     }
@@ -1891,7 +1929,24 @@ private Publisher<java.sql.Clob> convertClobBind(
           Mono.from(adapter.publishClobWrite(r2dbcClob.stream(), jdbcClob))
             .thenReturn(jdbcClob),
         jdbcClob -> {
-          addDeallocation(adapter.publishClobFree(jdbcClob));
+          Publisher<Void> freePublisher = adapter.publishClobFree(jdbcClob);
+
+          // Work around for Oracle JDBC bug #37160069: All LOBs need to be
+          // freed before closeAsyncOracle is called. This bug should be fixed
+          // by the 23.7 release of Oracle JDBC. The fix can be verified by the
+          // clobInsert and blobInsert methods in the TestKit class of the R2DBC
+          // SPI test: These tests will subscribe to Connection.close() before
+          // this freePublisher is subscribed to.
+          r2dbcConnection.addCloseTask(freePublisher);
+
+          addDeallocation(
+            Publishers.concatTerminal(
+              freePublisher,
+              Mono.fromRunnable(() ->
+                r2dbcConnection.removeCloseTask(freePublisher))));
+
+          // TODO: Why is discard() called here? It should be called by the
+          //  user who allocated the Clob, not by Oracle R2DBC.
           return r2dbcClob.discard();
         });
     }

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

@@ -1356,6 +1356,14 @@ public void testValidate() {
    */
   @Test
   public void testSetStatementTimeout() {
+    // Assume that oracle.jdbc.disablePipeline is only set to false when
+    // experimenting with pipelining on Mac OS. In this scenario, statement
+    // cancellation is known to not work.
+    String disabledProperty = System.getProperty("oracle.jdbc.disablePipeline");
+    assumeTrue(
+      disabledProperty == null || disabledProperty.equalsIgnoreCase("true"),
+      "oracle.jdbc.disablePipeline is set, and the value is not \"true\"");
+
     Connection connection =
       Mono.from(sharedConnection()).block(connectTimeout());
     try {

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

@@ -29,7 +29,10 @@
 import io.r2dbc.spi.Statement;
 import oracle.r2dbc.OracleR2dbcObject;
 import oracle.r2dbc.OracleR2dbcTypes;
+import oracle.r2dbc.test.DatabaseConfig;
+import oracle.r2dbc.test.TestUtils;
 import org.junit.jupiter.api.Test;
+import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
@@ -38,6 +41,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static java.util.Arrays.asList;

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

@@ -98,6 +98,7 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 /**
  * Verifies that
@@ -382,6 +383,14 @@ public void testConnectTimeout()
    */
   @Test
   public void testStatementTimeout() {
+    // Assume that oracle.jdbc.disablePipeline is only set to false when
+    // experimenting with pipelining on Mac OS. In this scenario, statement
+    // cancellation is known to not work.
+    String disabledProperty = System.getProperty("oracle.jdbc.disablePipeline");
+    assumeTrue(
+      disabledProperty == null || disabledProperty.equalsIgnoreCase("true"),
+      "oracle.jdbc.disablePipeline is set, and the value is not \"true\"");
+
     Connection connection0 =
       Mono.from(ConnectionFactories.get(connectionFactoryOptions()
         .mutate()

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

@@ -31,6 +31,7 @@
 import io.r2dbc.spi.Result;
 import io.r2dbc.spi.Result.Message;
 import io.r2dbc.spi.Result.UpdateCount;
+import io.r2dbc.spi.Row;
 import io.r2dbc.spi.Statement;
 import oracle.r2dbc.OracleR2dbcObject;
 import oracle.r2dbc.OracleR2dbcOptions;
@@ -76,12 +77,11 @@
 import static oracle.r2dbc.test.DatabaseConfig.connectionFactoryOptions;
 import static oracle.r2dbc.test.DatabaseConfig.databaseVersion;
 import static oracle.r2dbc.test.DatabaseConfig.jdbcMinorVersion;
-import static oracle.r2dbc.test.DatabaseConfig.jdbcVersion;
 import static oracle.r2dbc.test.DatabaseConfig.newConnection;
 import static oracle.r2dbc.test.DatabaseConfig.sharedConnection;
+import static oracle.r2dbc.test.DatabaseConfig.sqlTimeout;
 import static oracle.r2dbc.test.TestUtils.constructObject;
 import static oracle.r2dbc.test.TestUtils.showErrors;
-import static oracle.r2dbc.test.DatabaseConfig.sqlTimeout;
 import static oracle.r2dbc.util.Awaits.awaitError;
 import static oracle.r2dbc.util.Awaits.awaitExecution;
 import static oracle.r2dbc.util.Awaits.awaitMany;
@@ -3231,10 +3231,10 @@ public boolean equals(Object other) {
 
       // Oracle JDBC 23.4 has a defect which prevents the Subscriber from
       // receiving a terminal signal. The defect has been reported as bug
-      // #36607804, and is expected to be fixed in the 23.5 release.
+      // #36607804, it will be fixed in the 23.6 release.
       Assumptions.assumeTrue(
-        jdbcMinorVersion() >= 5,
-        "Oracle JDBC 23.4 does not support generated keys for VECTOR");
+        jdbcMinorVersion() >= 6,
+        "Oracle JDBC 23.5 does not support generated keys for VECTOR");
 
       IdVector expected2 = new IdVector(
         0,
@@ -3257,12 +3257,11 @@ public boolean equals(Object other) {
                 return Mono.just(((UpdateCount) segment).value());
               }
               else if (segment instanceof Result.RowSegment) {
-                OutParameters outParameters =
-                  ((Result.OutSegment)segment).outParameters();
+                Row generatedRow = ((Result.RowSegment)segment).row();
 
                 return Mono.just(new IdVector(
-                  outParameters.get("outId", Integer.class),
-                  outParameters.get("outVector", VECTOR.class)));
+                  generatedRow.get("id", Integer.class),
+                  generatedRow.get("value", VECTOR.class)));
               }
               else if (segment instanceof Message) {
                 throw ((Message)segment).exception();