BigQuery: BigQueryReadClient is not shutdown properly in Connection API

[naung9]

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

1. Service : BigQuery
2. OS type and version: MacOs Sonoma 14.7
3. Java version: 17
4. version(s): all versions

Steps to reproduce

1. Prepare select queries that runs longer than 2 minutes
2. 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)