PyFlink Table SQL Job on GKE Failed to Sink Data to BigQuery

[emersonwangsd]

Since flink-1.17-connector-bigquery-0.5.0.jar is compatible with Flink 1.20, PyFlink Table SQL job running on the Flink 1.20 GKE cluster failed to sink data to the associated BigQuerry table with the following errors in the task manager log files:

2025-02-21 06:22:16,933 WARN org.apache.flink.runtime.taskmanager.Task [] - Source: oregon_spprod_mms_residual_capacity[1] -> (Calc[2] -> StreamRecordTimestampInserter[3],
Calc[4] -> sink_table[5]: Writer) (1/1)#0 (87343a42f8f9da4bcf45973ede409d0b_e3dfc0d7e9ecd8a43f85f0b68ebf3b80_0_0) switched from RUNNING to FAILED with failure cause:
java.lang.Exception: Could not perform checkpoint 2 for operator Source: oregon_spprod_mms_residual_capacity[1] -> (Calc[2] -> StreamRecordTimestampInserter[3], Calc[4] -> sink_table[5]: Write
r) (1/1)#0.
at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointAsyncInMailbox(StreamTask.java:1322) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$triggerCheckpointAsync$17(StreamTask.java:1269) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:101) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:414) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:383) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:368) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949) [flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763) [flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) [flink-dist-1.20.0.jar:1.20.0]
at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: com.google.api.gax.rpc.DeadlineExceededException: io.grpc.StatusRuntimeException: DEADLINE_EXCEEDED: Deadline CallOptions will be exceeded in 599.971036571s.
at com.google.api.gax.rpc.ApiExceptionFactory.createException(ApiExceptionFactory.java:94) ~[gax-2.60.0.jar:2.60.0]
at com.google.api.gax.rpc.ApiExceptionFactory.createException(ApiExceptionFactory.java:41) ~[gax-2.60.0.jar:2.60.0]
at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:86) ~[gax-grpc-2.60.0.jar:2.60.0]
at com.google.api.gax.grpc.GrpcApiExceptionFactory.create(GrpcApiExceptionFactory.java:66) ~[gax-grpc-2.60.0.jar:2.60.0]
at com.google.api.gax.grpc.GrpcExceptionCallable$ExceptionTransformingFuture.onFailure(GrpcExceptionCallable.java:97) ~[gax-grpc-2.60.0.jar:2.60.0]
at com.google.api.core.ApiFutures$1.onFailure(ApiFutures.java:84) ~[api-common-2.44.0.jar:2.44.0]
at com.google.common.util.concurrent.Futures$CallbackListener.run(Futures.java:1127) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at com.google.common.util.concurrent.DirectExecutor.execute(DirectExecutor.java:31) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:1286) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:1055) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:807) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at io.grpc.stub.ClientCalls$GrpcFuture.setException(ClientCalls.java:651) ~[grpc-stub-1.70.0.jar:1.70.0]
at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:621) ~[grpc-stub-1.70.0.jar:1.70.0]
at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39) ~[grpc-api-1.70.0.jar:1.70.0]
at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23) ~[grpc-api-1.70.0.jar:1.70.0]
at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40) ~[grpc-api-1.70.0.jar:1.70.0]
at com.google.api.gax.grpc.ChannelPool$ReleasingClientCall$1.onClose(ChannelPool.java:569) ~[gax-grpc-2.60.0.jar:2.60.0]
at io.grpc.internal.DelayedClientCall$CloseListenerRunnable.runInContext(DelayedClientCall.java:432) ~[grpc-core-1.70.0.jar:1.70.0]
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[grpc-core-1.70.0.jar:1.70.0]
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[?:?]
... 1 more
Suppressed: com.google.api.gax.rpc.AsyncTaskException: Asynchronous task failed
at com.google.api.gax.rpc.ApiExceptions.callAndTranslateApiException(ApiExceptions.java:57) ~[gax-2.60.0.jar:2.60.0]
at com.google.api.gax.rpc.UnaryCallable.call(UnaryCallable.java:112) ~[gax-2.60.0.jar:2.60.0]
at com.google.cloud.bigquery.storage.v1.BigQueryWriteClient.getWriteStream(BigQueryWriteClient.java:557) ~[google-cloud-bigquerystorage-3.8.0.jar:3.8.0]
at com.google.cloud.bigquery.storage.v1.StreamWriter.lambda$new$0(StreamWriter.java:285) ~[google-cloud-bigquerystorage-3.8.0.jar:3.8.0]
at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(Unknown Source) ~[?:?]
at com.google.cloud.bigquery.storage.v1.StreamWriter.(StreamWriter.java:276) ~[google-cloud-bigquerystorage-3.8.0.jar:3.8.0]
at com.google.cloud.bigquery.storage.v1.StreamWriter.(StreamWriter.java:59) ~[google-cloud-bigquerystorage-3.8.0.jar:3.8.0]
at com.google.cloud.bigquery.storage.v1.StreamWriter$Builder.build(StreamWriter.java:896) ~[google-cloud-bigquerystorage-3.8.0.jar:3.8.0]
at com.google.cloud.flink.bigquery.services.BigQueryServicesImpl$StorageWriteClientImpl.createStreamWriter(BigQueryServicesImpl.java:289) ~[flink-connector-bigquery-common-0.5.
0.jar:0.5.0]
at com.google.cloud.flink.bigquery.sink.writer.BaseWriter.createStreamWriter(BaseWriter.java:256) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at com.google.cloud.flink.bigquery.sink.writer.BigQueryDefaultWriter.sendAppendRequest(BigQueryDefaultWriter.java:123) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at com.google.cloud.flink.bigquery.sink.writer.BaseWriter.append(BaseWriter.java:216) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at com.google.cloud.flink.bigquery.sink.writer.BaseWriter.flush(BaseWriter.java:145) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at com.google.cloud.flink.bigquery.sink.writer.BigQueryDefaultWriter.flush(BigQueryDefaultWriter.java:110) ~[flink-1.17-connector-bigquery-0.5.0.jar:0.5.0]
at org.apache.flink.streaming.runtime.operators.sink.SinkWriterOperator.prepareSnapshotPreBarrier(SinkWriterOperator.java:167) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.prepareSnapshotPreBarrier(RegularOperatorChain.java:89) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:332) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$18(StreamTask.java:1437) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1425) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointAsyncInMailbox(StreamTask.java:1310) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$triggerCheckpointAsync$17(StreamTask.java:1269) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:101) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMail(MailboxProcessor.java:414) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsWhenDefaultActionUnavailable(MailboxProcessor.java:383) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:368) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:229) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:973) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:917) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:970) ~[flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:949) [flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:763) [flink-dist-1.20.0.jar:1.20.0]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575) [flink-dist-1.20.0.jar:1.20.0]
at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: io.grpc.StatusRuntimeException: DEADLINE_EXCEEDED: Deadline CallOptions will be exceeded in 599.971036571s.
at io.grpc.Status.asRuntimeException(Status.java:532) ~[grpc-api-1.70.0.jar:1.70.0]
at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:621) ~[grpc-stub-1.70.0.jar:1.70.0]
at io.grpc.PartialForwardingClientCallListener.onClose(PartialForwardingClientCallListener.java:39) ~[grpc-api-1.70.0.jar:1.70.0]
at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:23) ~[grpc-api-1.70.0.jar:1.70.0]
at io.grpc.ForwardingClientCallListener$SimpleForwardingClientCallListener.onClose(ForwardingClientCallListener.java:40) ~[grpc-api-1.70.0.jar:1.70.0]
at com.google.api.gax.grpc.ChannelPool$ReleasingClientCall$1.onClose(ChannelPool.java:569) ~[gax-grpc-2.60.0.jar:2.60.0]
at io.grpc.internal.DelayedClientCall$CloseListenerRunnable.runInContext(DelayedClientCall.java:432) ~[grpc-core-1.70.0.jar:1.70.0]
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[grpc-core-1.70.0.jar:1.70.0]
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) ~[?:?]
... 1 more

The job could sink data to the GCS bucket without any issue. We'd appreciate it very much if you could please look into it and let's know the root causes.

Thanks,

Emerson