MODDATAIMP-1278: Fix broken RxJava transaction lifecycle in streaming methods

[okolawole-ebsco]

Purpose

Fix broken transaction lifecycle in the three streaming DAO methods (streamRecords, streamMarcRecordIds, streamSourceRecords) that caused connection pool leaks. The old code committed after rollback on error paths, double-closed connections, and silently discarded cleanup Futures.

return getCachedPool(tenantId)
  .rxGetConnection()
  .flatMapPublisher(conn -> conn.rxBegin()
    .flatMapPublisher(tx -> conn.rxPrepare(sql)
      .flatMapPublisher(pq -> pq.createStream(1)
        .toFlowable()
        .map(this::toRow)
        .map(this::toRecord))
      .doAfterTerminate(tx::commit)       // A: fires on BOTH success AND error
      .doOnError(error -> {               // B: rollback + close
          tx.rollback();
          conn.close();
      })
      .doFinally(conn::close)));          // C: close connection again

Approach

A simple fix could look like so:

return getCachedPool(tenantId)
  .rxGetConnection()
  .flatMapPublisher(conn -> conn.rxBegin()
    .flatMapPublisher(tx -> conn.rxPrepare(sql)
      .flatMapPublisher(pq -> pq.createStream(1)
        .toFlowable()
        .map(this::toRow)
        .map(this::toRecord))
      .doOnComplete(() -> tx.commit())
      .doOnError(error -> tx.rollback())
      .doFinally(conn::close)));

But its missing the following:

• tx.commit()andtx.rollback()` still return Futures that are silently discarded. If commit fails, the error is lost.
• It never closes the PreparedStatement (pq).
• When a subscriber cancels mid-stream (e.g., firstOrError(), take(n)), neither doOnComplete nor doOnError fires. Only doFinally fires. So the transaction is never explicitly committed or rolled back.
• The stream cursor should be closed before rollback to avoid issuing rollback while a fetch is in-flight.

For a full solution, consolidate the three duplicated streaming patterns into a shared streamInTransaction() helper with a StreamContext inner class. The new lifecycle ensures: commit only on success, rollback only on error/cancel, stream cursor quiesced before rollback, exactly-once cleanup via AtomicBoolean guard, and cancel-path errors logged instead of swallowed. Includes a test that exercises 25 rapid cancel cycles to prove the pool is not exhausted.

Refs: https://folio-org.atlassian.net/browse/MODDATAIMP-1278

[sonarqubecloud]

Quality Gate passed

Issues
0 New issues
0 Accepted issues

Measures
0 Security Hotspots
83.3% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud