CloudPubSubSourceTask(StreamingPull) blocks indefinitely during poll() causing graceful shutdown failures

[Lance726]

Environment details

1. Java version: OpenJDK 18
2. Connector version: latest
3. Kafka version: 3.x or latest

Steps to reproduce

1. Configure and start a Kafka connector using CloudPubSubSourceTask(StreamingPull)
2. Attempt to stop the connector or restart the service
3. Observe that the task exceeds the task.shutdown.graceful.timeout.ms timeout and is forcibly cancelled
4. Observe numerous error logs being generated during shutdown or restart

Code example

// In CloudPubSubSourceTask#poll()
// StreamingPull
List<ReceivedMessage> response = subscriber.pull().get(); // This blocks indefinitely

Stack trace

Graceful stop of task xxxx failed.

SEVERE: *~*~*~ Previous channel ManagedChannelImpl{logId=1, target=pubsub.googleapis.com:443} was garbage collected without being shut down! ~*~*~*
Any relevant stacktrace here.


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:709)
	at io.grpc.ForwardingChannelBuilder2.build(ForwardingChannelBuilder2.java:272)
	at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createSingleChannel(InstantiatingGrpcChannelProvider.java:497)
	at com.google.api.gax.grpc.ChannelPool.<init>(ChannelPool.java:106)
	at com.google.api.gax.grpc.ChannelPool.create(ChannelPool.java:84)
	at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createChannel(InstantiatingGrpcChannelProvider.java:267)
	at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.getTransportChannel(InstantiatingGrpcChannelProvider.java:260)
	at com.google.api.gax.rpc.ClientContext.create(ClientContext.java:225)
	at com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub.create(GrpcSubscriberStub.java:287)
	at com.google.pubsub.kafka.source.CloudPubSubSourceConnector.verifySubscription(CloudPubSubSourceConnector.java:305)
	at com.google.pubsub.kafka.source.CloudPubSubSourceConnector.start(CloudPubSubSourceConnector.java:146)
	at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:216)
	at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:245)
	at org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:404)
	at org.apache.kafka.connect.runtime.WorkerConnector.doTransitionTo(WorkerConnector.java:385)
	at org.apache.kafka.connect.runtime.WorkerConnector.doRun(WorkerConnector.java:165)
	at org.apache.kafka.connect.runtime.WorkerConnector.run(WorkerConnector.java:125)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:238)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
	at java.base/java.util.concurrent.FutureTask.run(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.base/java.lang.Thread.run(Unknown Source)

External references such as API reference guides

• Kafka Connect source task execution loop: https://github.com/apache/kafka/blob/75f384a0199a5eded947defa428b8e9786cee726/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractWorkerSourceTask.java#L346-L363
  When a task is stopped, the while loop's condition needs to be evaluated to properly exit the loop, but this can't happen if the thread is blocked in poll()

• Kafka Connect shutdown latch release: https://github.com/apache/kafka/blob/75f384a0199a5eded947defa428b8e9786cee726/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java#L286-L303
  The shutdown latch is only released after the execution loop exits

• Kafka Connect graceful shutdown mechanism: https://github.com/apache/kafka/blob/75f384a0199a5eded947defa428b8e9786cee726/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerTask.java#L169-L175
  If the shutdown latch isn't released in time, the task is forcibly terminated without proper cleanup

Any additional information below

In the current implementation of CloudPubSubSourceTask#poll(), the call to subscriber.pull().get() blocks indefinitely while waiting for messages. This blocking behavior prevents proper shutdown of the connector:

1. When a connector shutdown is initiated, the task's main loop cannot exit because the thread is blocked in poll()
2. This causes the task to exceed the task.shutdown.graceful.timeout.ms timeout
3. As a result, the task is forcibly terminated rather than gracefully shut down
4. The CloudPubSubSourceTask#stop() method is never called
5. Critical resources like the PubSub subscriber are not properly closed

The error logs clearly demonstrate this issue:

1. "Graceful stop of task xxxx failed" indicates the timeout during shutdown
2. "SEVERE: ~~*~ Previous channel ManagedChannelImpl{logId=1, target=pubsub.googleapis.com:443} was garbage collected without being shut down!" confirms that gRPC channels are being leaked

Suggested solution: The pull() operation should either implement a reasonable timeout or be made interruptible, allowing the task to properly respond to shutdown requests and clean up resources within the graceful shutdown period.

If there are any specific considerations behind the current implementation, please let me know. Thanks!