BigQuery: BigQueryReadClient is not shutdown properly in Connection API #3571
Description
When long running queries ( > 2 minutes) are executed concurrently (in my case 10 concurrent queries) are executed using BigQuery Connection Api, queries are executed successfully but gRPC ManagedChannelImpl is producing SEVERE log
*~*~*~ Previous channel ManagedChannelImpl{logId=1085, target=bigquerystorage.googleapis.com:443} was not shutdown properly!!! ~*~*~*
Make sure to call shutdown()/shutdownNow() and wait until awaitTermination() returns true
After investigation, it is most likely caused by not closing/shutting down bqReadClient in ConnectionImpl.
Environment details
- Service : BigQuery
- OS type and version: MacOs Sonoma 14.7
- Java version: 17
- version(s): all versions
Steps to reproduce
- Prepare select queries that runs longer than 2 minutes
- Run the queries concurrently
Code example
public void test() {
BigQuery bq = BigQueryOptions.getDefaultInstance().getService();
List<String> queries = List.of(longRunQuery1, longRunQuery2,...,longRunQuery10);
try (ScheduledExecutorService executor = Executors.newScheduledThreadPool(10)) {
for (String query : queries) {
executor.submit(() -> {
Connection connection = bq.createConnection();
try {
connection.executeSelect(query);
} catch (BigQuerySQLException e) {
e.printStackTrace();
}
});
}
}
}Stack trace
*~*~*~ Previous channel ManagedChannelImpl{logId=153, target=bigquerystorage.googleapis.com:443} was not shutdown properly!!! ~*~*~*
Make sure to call shutdown()/shutdownNow() and wait until awaitTermination() returns true.: java.lang.RuntimeException: ManagedChannel allocation site
at io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init>(ManagedChannelOrphanWrapper.java:102)
at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:60)
at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:51)
at io.grpc.internal.ManagedChannelImplBuilder.build(ManagedChannelImplBuilder.java:655)
at io.grpc.ForwardingChannelBuilder2.build(ForwardingChannelBuilder2.java:261)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createSingleChannel(InstantiatingGrpcChannelProvider.java:442)
at com.google.api.gax.grpc.ChannelPool.<init>(ChannelPool.java:107)
at com.google.api.gax.grpc.ChannelPool.create(ChannelPool.java:85)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createChannel(InstantiatingGrpcChannelProvider.java:243)
at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.getTransportChannel(InstantiatingGrpcChannelProvider.java:237)
at com.google.api.gax.rpc.ClientContext.create(ClientContext.java:226)
at com.google.cloud.bigquery.storage.v1.stub.EnhancedBigQueryReadStub.create(EnhancedBigQueryReadStub.java:100)
at com.google.cloud.bigquery.storage.v1.BigQueryReadClient.<init>(BigQueryReadClient.java:130)
at com.google.cloud.bigquery.storage.v1.BigQueryReadClient.create(BigQueryReadClient.java:110)
at com.google.cloud.bigquery.storage.v1.BigQueryReadClient.create(BigQueryReadClient.java:102)
at com.google.cloud.bigquery.ConnectionImpl.highThroughPutRead(ConnectionImpl.java:954)
at com.google.cloud.bigquery.ConnectionImpl.getSubsequentQueryResultsWithJob(ConnectionImpl.java:848)
at com.google.cloud.bigquery.ConnectionImpl.queryRpc(ConnectionImpl.java:489)
at com.google.cloud.bigquery.ConnectionImpl.getExecuteSelectResponse(ConnectionImpl.java:239)
at com.google.cloud.bigquery.ConnectionImpl.executeSelect(ConnectionImpl.java:198)
at com.rakuten.neogaruda.query.repository.bigquery.SubscriberRepositoryImpl.lambda$getSubscribersByQuery$1(SubscriberRepositoryImpl.java:221)
at io.smallrye.context.impl.wrappers.SlowContextualConsumer.accept(SlowContextualConsumer.java:21)
at io.smallrye.mutiny.operators.multi.builders.EmitterBasedMulti.subscribe(EmitterBasedMulti.java:65)
at io.smallrye.mutiny.operators.AbstractMulti.subscribe(AbstractMulti.java:39)
at io.smallrye.mutiny.operators.multi.MultiFlatMapOp$FlatMapMainSubscriber.onItem(MultiFlatMapOp.java:182)
at io.smallrye.mutiny.subscription.MultiSubscriber.onNext(MultiSubscriber.java:61)
at io.smallrye.mutiny.converters.uni.UniToMultiPublisher$UniToMultiSubscription.onItem(UniToMultiPublisher.java:94)
at io.smallrye.mutiny.operators.uni.UniOperatorProcessor.onItem(UniOperatorProcessor.java:47)
at io.smallrye.mutiny.operators.uni.builders.UniCreateFromItemSupplier.subscribe(UniCreateFromItemSupplier.java:29)
at io.smallrye.mutiny.operators.AbstractUni.subscribe(AbstractUni.java:36)
at io.smallrye.mutiny.operators.uni.UniRunSubscribeOn.lambda$subscribe$0(UniRunSubscribeOn.java:27)
at io.quarkus.vertx.core.runtime.VertxCoreRecorder$14.runWith(VertxCoreRecorder.java:587)
at org.jboss.threads.EnhancedQueueExecutor$Task.run(EnhancedQueueExecutor.java:2513)
at org.jboss.threads.EnhancedQueueExecutor$ThreadBody.run(EnhancedQueueExecutor.java:1538)
at org.jboss.threads.DelegatingRunnable.run(DelegatingRunnable.java:29)
at org.jboss.threads.ThreadLocalResettingRunnable.run(ThreadLocalResettingRunnable.java:29)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:840)