Hi,
I've tested the Flink SQL-gateway to run some simple Hive queries but met
some exceptions.
Environment Description:
Run on : Kubernetes
Deployment Mode: Session Mode (created by a flink-kubernetes-operator)
Steps to run:
1. Apply a `flinkdeployment` of flink session cluster to flink operator
```
apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
name: flink-session-cluster-example
namespace: xxx
spec:
image: xxx/flink:1.17-sql-gateway-dev
flinkVersion: v1_17
flinkConfiguration:
taskmanager.numberOfTaskSlots: "2"
pipeline.max-parallelism: "1000"
state.backend.type: rocksdb
state.backend.incremental: "true"
state.checkpoints.dir: xxx
execution.checkpointing.interval: 1m
execution.checkpointing.timeout: 30m
high-availability:
org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
high-availability.storageDir: xxx
akka.framesize: 20971520b
execution.checkpointing.externalized-checkpoint-retention:
RETAIN_ON_CANCELLATION
taskmanager.memory.managed.fraction: "0.2"
kubernetes.hadoop.conf.config-map.name: xxx
serviceAccount: default
podTemplate:
apiVersion: v1
kind: Pod
metadata:
name: pod-template
spec:
serviceAccount: default
jobManager:
resource:
memory: "2048m"
cpu: 1
taskManager:
resource:
memory: "4096m"
cpu: 1
```
This image has been built with a `hadoop dependency` , an existing `hadoop
configmap`.
2. Login to the job-manager pod and run the followings
`./bin/sql-gateway.sh start-foreground
-Dsql-gateway.endpoint.type=hiveserver2
-Dsql-gateway.endpoint.hiveserver2.catalog.hive-conf-dir=/opt/flink/hive-conf`
3. Start a beeline and connect to the SQL gateway then run a simple Hive
query
`select count(1) from simple_demo_output where dt = '2021-08-14';`
4.The SQL gateway goes wrong with the following logs:
```
2023-07-03 06:27:11,078 INFO
org.apache.flink.client.program.rest.RestClusterClient
[] - Submitting job 'collect' (4c99c40392cb935d3df94891655d2ce5).
2023-07-03 06:27:15,092 INFO
org.apache.flink.client.program.rest.RestClusterClient
[] - Successfully submitted job 'collect'
(4c99c40392cb935d3df94891655d2ce5) to '
http://flink-session-cluster-example-rest.realtime-streaming:8081'.
2023-07-03 06:27:15,879 ERROR
org.apache.flink.table.gateway.service.operation.OperationManager [] -
Failed to execute the operation 7613f663-8641-428c-b3d2-ec77a12fa6ee.
org.apache.flink.table.api.TableException: Failed to execute sql
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeQueryOperation(TableEnvironmentImpl.java:974)
~[flink-table-api-java-uber-1.17.1.jar:1.17.1]
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:1422)
~[flink-table-api-java-uber-1.17.1.jar:1.17.1]
at
org.apache.flink.table.gateway.service.operation.OperationExecutor.executeOperation(OperationExecutor.java:431)
~[flink-sql-gateway-1.17.1.jar:1.17.1]
at
org.apache.flink.table.gateway.service.operation.OperationExecutor.executeStatement(OperationExecutor.java:195)
~[flink-sql-gateway-1.17.1.jar:1.17.1]
at
org.apache.flink.table.gateway.service.SqlGatewayServiceImpl.lambda$executeStatement$1(SqlGatewayServiceImpl.java:212)
~[flink-sql-gateway-1.17.1.jar:1.17.1]
at
org.apache.flink.table.gateway.service.operation.OperationManager.lambda$submitOperation$1(OperationManager.java:119)
~[flink-sql-gateway-1.17.1.jar:1.17.1]
at
org.apache.flink.table.gateway.service.operation.OperationManager$Operation.lambda$run$0(OperationManager.java:258)
~[flink-sql-gateway-1.17.1.jar:1.17.1]
at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) [?:?]
at java.util.concurrent.FutureTask.run(Unknown Source) [?:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) [?:?]
at java.util.concurrent.FutureTask.run(Unknown Source) [?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source) [?:?]
at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: org.apache.flink.util.FlinkException: Failed to execute job
'collect'.
at
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:2212)
~[flink-dist-1.17.1.jar:1.17.1]
at
org.apache.flink.table.planner.delegation.DefaultExecutor.executeAsync(DefaultExecutor.java:95)
~[flink-table-planner_2.12-1.17.1.jar:1.17.1]
at
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeQueryOperation(TableEnvironmentImpl.java:955)
~[flink-table-api-java-uber-1.17.1.jar:1.17.1]
... 13 more
Caused by: java.lang.RuntimeException: Error while waiting for job to be
initialized
at
org.apache.flink.client.ClientUtils.waitUntilJobInitializationFinished(ClientUtils.java:151)
~[flink-dist-1.17.1.jar:1.17.1]
at
org.apache.flink.client.deployment.executors.AbstractSessionClusterExecutor.lambda$execute$2(AbstractSessionClusterExecutor.java:87)
~[flink-dist-1.17.1.jar:1.17.1]
at
org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:73)
~[flink-dist-1.17.1.jar:1.17.1]
at java.util.concurrent.CompletableFuture$UniApply.tryFire(Unknown Source)
~[?:?]
at java.util.concurrent.CompletableFuture$Completion.run(Unknown Source)
~[?:?]
... 1 more
Caused by: java.util.concurrent.ExecutionException:
org.apache.flink.runtime.rest.util.RestClientException: [Internal server
error., <Exception on server side:
org.apache.flink.runtime.rpc.exceptions.RecipientUnreachableException:
Could not send message
[LocalFencedMessage(863401565fbee9bb5587a17d5279473a,
LocalRpcInvocation(JobMasterGateway.requestJob(Time)))] from sender
[Actor[akka://flink/temp/jobmanager_2$4d]] to recipient
[Actor[akka://flink/user/rpc/jobmanager_2#-1135810317]], because the
recipient is unreachable. This can either mean that the recipient has been
terminated or that the remote RpcService is currently not reachable.
at
org.apache.flink.runtime.rpc.akka.DeadLettersActor.handleDeadLetter(DeadLettersActor.java:61)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20)
at scala.PartialFunction.applyOrElse(PartialFunction.scala:127)
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126)
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175)
at akka.actor.Actor.aroundReceive(Actor.scala:537)
at akka.actor.Actor.aroundReceive$(Actor.scala:535)
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:579)
at akka.actor.ActorCell.invoke(ActorCell.scala:547)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
at akka.dispatch.Mailbox.run(Mailbox.scala:231)
at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
at
java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown
Source)
at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source)
at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source)
End of exception on server side>]
at java.util.concurrent.CompletableFuture.reportGet(Unknown Source) ~[?:?]
at java.util.concurrent.CompletableFuture.get(Unknown Source) ~[?:?]
at
org.apache.flink.client.deployment.executors.AbstractSessionClusterExecutor.lambda$null$0(AbstractSessionClusterExecutor.java:88)
~[flink-dist-1.17.1.jar:1.17.1]
at
org.apache.flink.client.ClientUtils.waitUntilJobInitializationFinished(ClientUtils.java:135)
~[flink-dist-1.17.1.jar:1.17.1]
at
org.apache.flink.client.deployment.executors.AbstractSessionClusterExecutor.lambda$execute$2(AbstractSessionClusterExecutor.java:87)
~[flink-dist-1.17.1.jar:1.17.1]
at
org.apache.flink.util.function.FunctionUtils.lambda$uncheckedFunction$2(FunctionUtils.java:73)
~[flink-dist-1.17.1.jar:1.17.1]
at java.util.concurrent.CompletableFuture$UniApply.tryFire(Unknown Source)
~[?:?]
at java.util.concurrent.CompletableFuture$Completion.run(Unknown Source)
~[?:?]
... 1 more
Caused by: org.apache.flink.runtime.rest.util.RestClientException:
[Internal server error., <Exception on server side:
org.apache.flink.runtime.rpc.exceptions.RecipientUnreachableException:
Could not send message
[LocalFencedMessage(863401565fbee9bb5587a17d5279473a,
LocalRpcInvocation(JobMasterGateway.requestJob(Time)))] from sender
[Actor[akka://flink/temp/jobmanager_2$4d]] to recipient
[Actor[akka://flink/user/rpc/jobmanager_2#-1135810317]], because the
recipient is unreachable. This can either mean that the recipient has been
terminated or that the remote RpcService is currently not reachable.
at
org.apache.flink.runtime.rpc.akka.DeadLettersActor.handleDeadLetter(DeadLettersActor.java:61)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24)
at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20)
at scala.PartialFunction.applyOrElse(PartialFunction.scala:127)
at scala.PartialFunction.applyOrElse$(PartialFunction.scala:126)
at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20)
at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:175)
at akka.actor.Actor.aroundReceive(Actor.scala:537)
at akka.actor.Actor.aroundReceive$(Actor.scala:535)
at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220)
at akka.actor.ActorCell.receiveMessage(ActorCell.scala:579)
at akka.actor.ActorCell.invoke(ActorCell.scala:547)
at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
at akka.dispatch.Mailbox.run(Mailbox.scala:231)
at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
at java.base/java.util.concurrent.ForkJoinTask.doExec(Unknown Source)
at
java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(Unknown
Source)
at java.base/java.util.concurrent.ForkJoinPool.scan(Unknown Source)
at java.base/java.util.concurrent.ForkJoinPool.runWorker(Unknown Source)
at java.base/java.util.concurrent.ForkJoinWorkerThread.run(Unknown Source)
End of exception on server side>]
at
org.apache.flink.runtime.rest.RestClient.parseResponse(RestClient.java:536)
~[flink-dist-1.17.1.jar:1.17.1]
at
org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$3(RestClient.java:516)
~[flink-dist-1.17.1.jar:1.17.1]
at java.util.concurrent.CompletableFuture$UniCompose.tryFire(Unknown
Source) ~[?:?]
at java.util.concurrent.CompletableFuture$Completion.run(Unknown Source)
~[?:?]
... 3 more
command terminated with exit code 137
```
And then the `flinkdeployment` goes wrong and cannot recover itself
