BigQuery Java Connection API can hang / deadlock if connection is closed without reading all results

[jonathanswenson]

Thanks for stopping by to let us know something could be better!

If you are still having issues, please include as much information as possible:

Environment details

1. Specify the API at the beginning of the title. For example, "BigQuery: ...").
   General, Core, and Other are also allowed as types
2. OS type and version: Linux Ubuntu
3. Java version: 17 adoptium
4. version(s): 2.22.0

Steps to reproduce

1. Run query that returns enough rows to fill the async buffer
2. wait for the query to complete,
3. read some data via JDBC API (but not all of it)
4. ensure that the buffer is full (can just sleep)
5. close connection

Code example

ConnectionSettings connectionSettings =
ConnectionSettings.newBuilder()
       .setUseReadAPI(false)
       .setRequestTimeout(10L)
       .setMaxResults(100L)
       .setUseQueryCache(true)
       .build();
Connection connection = bigquery.createConnection(connectionSettings);
String selectQuery = "<query that returns more than buffer size rows>";
ListenableFuture<ExecuteSelectResponse> executeSelectFuture = connection.executeSelectAsync(selectQuery, ...);

ExecuteSelectResponse response = executeSelectFuture.get();
ResultSet results = response.getResultSet();

// read some, but not all of the data
resultSet.next()
// sleep for a bit to ensure that the buffer is full
Thread.sleep(10000); 

// hangs forever.
connection.close();

Stack trace (from jstack thread dump)

"thread name" #35 prio=5 os_prio=0 cpu=917.01ms elapsed=4251.27s tid=0x0000558529454f40 nid=0x49 waiting on condition  [0x00007f9eefffd000]
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park(java.base@17.0.6/Native Method)
	- parking to wait for  <0x000000074b91f798> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
	at java.util.concurrent.locks.LockSupport.park(java.base@17.0.6/LockSupport.java:341)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionNode.block(java.base@17.0.6/AbstractQueuedSynchronizer.java:506)
	at java.util.concurrent.ForkJoinPool.unmanagedBlock(java.base@17.0.6/ForkJoinPool.java:3463)
	at java.util.concurrent.ForkJoinPool.managedBlock(java.base@17.0.6/ForkJoinPool.java:3434)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(java.base@17.0.6/AbstractQueuedSynchronizer.java:1623)
	at java.util.concurrent.LinkedBlockingDeque.putLast(java.base@17.0.6/LinkedBlockingDeque.java:389)
	at java.util.concurrent.LinkedBlockingDeque.put(java.base@17.0.6/LinkedBlockingDeque.java:642)
	at com.google.cloud.bigquery.ConnectionImpl.flagEndOfStream(ConnectionImpl.java:767)
	at com.google.cloud.bigquery.ConnectionImpl.close(ConnectionImpl.java:144)
	- locked <0x0000000743dab650> (a com.google.cloud.bigquery.ConnectionImpl)
	at ... my code calling close

Can see that the thread is blocked here: https://github.com/googleapis/java-bigquery/blob/main/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java#L767-L768

Async worker thread that is populating the queue is also blocked (as the queue is full).

"pool-67-thread-3" #239 prio=5 os_prio=0 cpu=186.77ms elapsed=8312.48s tid=0x00007f9ef4010560 nid=0x12c waiting on condition  [0x00007f9ee81fe000]
   java.lang.Thread.State: WAITING (parking)
	at jdk.internal.misc.Unsafe.park(java.base@17.0.6/Native Method)
	- parking to wait for  <0x000000074b91f798> (a java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject)
	at java.util.concurrent.locks.LockSupport.park(java.base@17.0.6/LockSupport.java:341)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionNode.block(java.base@17.0.6/AbstractQueuedSynchronizer.java:506)
	at java.util.concurrent.ForkJoinPool.unmanagedBlock(java.base@17.0.6/ForkJoinPool.java:3463)
	at java.util.concurrent.ForkJoinPool.managedBlock(java.base@17.0.6/ForkJoinPool.java:3434)
	at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(java.base@17.0.6/AbstractQueuedSynchronizer.java:1623)
	at java.util.concurrent.LinkedBlockingDeque.putLast(java.base@17.0.6/LinkedBlockingDeque.java:389)
	at java.util.concurrent.LinkedBlockingDeque.put(java.base@17.0.6/LinkedBlockingDeque.java:642)
	at com.google.cloud.bigquery.ConnectionImpl.lambda$populateBufferAsync$4(ConnectionImpl.java:733)
	at com.google.cloud.bigquery.ConnectionImpl$$Lambda$1385/0x0000000801740228.run(Unknown Source)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(java.base@17.0.6/ThreadPoolExecutor.java:1136)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(java.base@17.0.6/ThreadPoolExecutor.java:635)
	at java.lang.Thread.run(java.base@17.0.6/Thread.java:833)