fix(jdbc-driver): streamQuery connection release (#6108)

Author: buntarb

packages/cubejs-jdbc-driver/src/JDBCDriver.ts

@@ -225,58 +225,46 @@ export class JDBCDriver extends BaseDriver {
   }
 
   public async streamQuery(sql: string, values: string[]): Promise<Readable> {
+    const conn = await this.pool.acquire();
     const query = applyParams(sql, values);
     const cancelObj: {cancel?: Function} = {};
     try {
-      const conn = await this.pool.acquire();
-      try {
-        const createStatement = promisify(conn.createStatement.bind(conn));
-        const statement = await createStatement();
+      const createStatement = promisify(conn.createStatement.bind(conn));
+      const statement = await createStatement();
 
-        // TODO (buntarb): this does not make any sense...
-        // statement.setFetchSize(
-        //   getEnv('dbQueryStreamHighWaterMark'),
-        //   (err: unknown) => { if (err) console.error(err); }
-        // );
-
-        if (cancelObj) {
-          cancelObj.cancel = promisify(statement.cancel.bind(statement));
-        }
+      if (cancelObj) {
+        cancelObj.cancel = promisify(statement.cancel.bind(statement));
+      }
 
-        // TODO (buntarb): timeout decision needs.
-        // const setQueryTimeout = promisify(statement.setQueryTimeout.bind(statement));
-        // await setQueryTimeout(600);
-
-        const executeQuery = promisify(statement.execute.bind(statement));
-        const resultSet = await executeQuery(query);
-        return new Promise((resolve, reject) => {
-          resultSet.toObjectIter(
-            (
-              err: unknown,
-              res: {
+      const executeQuery = promisify(statement.execute.bind(statement));
+      const resultSet = await executeQuery(query);
+      return new Promise((resolve, reject) => {
+        resultSet.toObjectIter(
+          (
+            err: unknown,
+            res: {
                 labels: string[],
                 types: number[],
                 rows: { next: nextFn },
               },
-            ) => {
-              if (err) reject(err);
-              const rowsStream = new QueryStream(res.rows.next);
-              const cleanup = (e?: Error) => {
-                if (!rowsStream.destroyed) {
-                  rowsStream.destroy(e);
-                }
-              };
-              rowsStream.once('end', cleanup);
-              rowsStream.once('error', cleanup);
-              rowsStream.once('close', cleanup);
-              resolve(rowsStream);
-            }
-          );
-        });
-      } finally {
-        await this.pool.release(conn);
-      }
+          ) => {
+            if (err) reject(err);
+            const rowsStream = new QueryStream(res.rows.next);
+            const cleanup = (e?: Error) => {
+              if (!rowsStream.destroyed) {
+                this.pool.release(conn);
+                rowsStream.destroy(e);
+              }
+            };
+            rowsStream.once('end', cleanup);
+            rowsStream.once('error', cleanup);
+            rowsStream.once('close', cleanup);
+            resolve(rowsStream);
+          }
+        );
+      });
     } catch (ex: any) {
+      await this.pool.release(conn);
       if (ex.cause) {
         throw new Error(ex.cause.getMessageSync());
       } else {