https://flink.apache.org/ logo
Join Slack
Powered by
Hello folks, I am experimenting with Flink and hav...
# troubleshooting
a
Hello folks, I am experimenting with Flink and have been trying to get a setup for SQL Client -> SQL Gateway -> Flink. I got it to work before but somehow I managed to break it and I'm a bit lost now. This is what I do, (I use minikube):
Copy code
kubectl create clusterrolebinding flink-role-binding-default --clusterrole=edit --serviceaccount=default:default
$FLINK_PATH/bin/kubernetes-session.sh -Dkubernetes.cluster-id=flink-session-cluster -Dkubernetes.rest-service.exposed.type=LoadBalancer -Dkubernetes.namespace=default -Drest.address=0.0.0.0
This works fine and creates the cluster on minikube, I can access it if I do 
minikube tunnel
to access the UI. At this point my services look like this:
Copy code
βœ— kg svc
NAME                         TYPE           CLUSTER-IP      EXTERNAL-IP   PORT(S)             AGE
flink-session-cluster        ClusterIP      None            <none>        6123/TCP,6124/TCP   5s
flink-session-cluster-rest   LoadBalancer   10.100.202.72   127.0.0.1     8081:32409/TCP      5s
kubernetes                   ClusterIP      10.96.0.1       <none>        443/TCP             4d18h
I then create the SQL Gateway with the below yaml file:
Copy code
apiVersion: apps/v1
kind: Deployment
metadata:
  name: sql-gateway
spec:
  replicas: 1
  selector:
    matchLabels:
      app: sql-gateway
  template:
    metadata:
      labels:
        app: sql-gateway
    spec:
      containers:
        - name: sql-gateway
          image: flink:1.17.0
          ports:
            - containerPort: 8083
          command: ["/bin/sh", "-c"]
          args:
            - "/opt/flink/bin/sql-gateway.sh start-foreground"
          volumeMounts:
            - name: flink-conf
              mountPath: /opt/flink/conf/flink-conf.yaml
      volumes:
        - name: flink-conf
          hostPath:
            path: /host/flink-refinery/k8s/conf/flink-conf.yaml  # minikube stuff
            type: File
---
apiVersion: v1
kind: Service
metadata:
  name: sql-gateway
spec:
  selector:
    app: sql-gateway
  ports:
    - name: sql-gateway
      port: 8083
      targetPort: 8083
  type: ClusterIP
And my 
flink-conf.yaml
is
Copy code
jobmanager.rpc.address: flink-session-cluster
jobmanager.rpc.port: 6123
rest.address: flink-session-cluster-rest
rest.port: 8081
sql-gateway.endpoint.rest.address: 0.0.0.0
sql-gateway.endpoint.rest.port: 8083
sql-gateway.endpoint.type: rest
I can confirm from the logs that this config is picked up.
Copy code
2023-03-29 18:33:38,431 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: rest.port, 8081
2023-03-29 18:33:38,431 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: sql-gateway.endpoint.rest.address, sql-gateway
2023-03-29 18:33:38,431 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: sql-gateway.endpoint.type, rest
2023-03-29 18:33:38,431 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.rpc.address, flink-session-cluster
2023-03-29 18:33:38,431 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: sql-gateway.endpoint.rest.port, 8083
2023-03-29 18:33:38,432 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: rest.address, flink-session-cluster-rest
2023-03-29 18:33:38,432 INFO  org.apache.flink.configuration.GlobalConfiguration           [] - Loading configuration property: jobmanager.rpc.port, 6123
When I create a new Flink pod and run the sql client, however, I get an error from SQL Gateway. These are my pods and services at this point:
Copy code
βœ— kg po
NAME                                     READY   STATUS    RESTARTS   AGE
flink-session-cluster-758bc469cc-lgpps   1/1     Running   0          10m
sql-client-74f44556cf-ls24w              1/1     Running   0          2m29s
sql-gateway-7758c8fdf9-nldvt             1/1     Running   0          2m29s
βœ— kg svc
NAME                         TYPE           CLUSTER-IP       EXTERNAL-IP   PORT(S)             AGE
flink-session-cluster        ClusterIP      None             <none>        6123/TCP,6124/TCP   10m
flink-session-cluster-rest   LoadBalancer   10.100.202.72    127.0.0.1     8081:32409/TCP      10m
kubernetes                   ClusterIP      10.96.0.1        <none>        443/TCP             4d18h
sql-gateway                  ClusterIP      10.103.121.161   <none>        8083/TCP            2m30s
And I use sql-client pod to connect to the gateway like this:
Copy code
βœ— kubectl exec -i -t sql-client-74f44556cf-ls24w -- /bin/bash
root@sql-client-74f44556cf-5tv5d:/opt/flink# ./bin/sql-client.sh gateway -e sql-gateway:8083
# At this point I get "Session <id> is opened, and the number..." from SQL Gateway logs, so I can connect to the gateway
Flink SQL> SELECT
>   name,
>   COUNT(*) AS cnt
> FROM
>   (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name)
> GROUP BY name;
As soon as I run that test SQL command, I get this error from the gateway's logs:
Copy code
2023-03-29 18:32:28,748 WARN  org.apache.flink.client.program.rest.RestClusterClient       [] - Attempt to submit job 'collect' (ccaafce00ee8c95e4b5c037774d885e1) to '<http://0.0.0.0:8081>' has failed.
java.util.concurrent.CompletionException: org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /0.0.0.0:8081
	at java.util.concurrent.CompletableFuture.encodeThrowable(Unknown Source) ~[?:?]
	at java.util.concurrent.CompletableFuture.completeThrowable(Unknown Source) ~[?:?]
	at java.util.concurrent.CompletableFuture$UniCompose.tryFire(Unknown Source) ~[?:?]
	at java.util.concurrent.CompletableFuture.postComplete(Unknown Source) ~[?:?]
	at java.util.concurrent.CompletableFuture.completeExceptionally(Unknown Source) ~[?:?]
	at org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$1(RestClient.java:480) ~[flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:578) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:571) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:550) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:491) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:616) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:609) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:117) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:321) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:337) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:776) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) [flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [flink-dist-1.17.0.jar:1.17.0]
	at java.lang.Thread.run(Unknown Source) [?:?]
Caused by: org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: /0.0.0.0:8081
Caused by: java.net.ConnectException: Connection refused
	at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[?:?]
	at sun.nio.ch.SocketChannelImpl.finishConnect(Unknown Source) ~[?:?]
	at org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:337) ~[flink-dist-1.17.0.jar:1.17.0]
	at org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:334) ~[flink-dist-1.17.0.jar:1.17.0]
Now I am extremely confused, why it's trying to connect to 
0.0.0.0:8081
? I remember I got this to work at some point, but I'm not sure what was different since I was just quickly experimenting.. If anyone has any pointers, I'd really appreciate it! Thanks!
To follow up on this, should I run 
sql-gateway
on the same pod as the job manager? Is that the assumption?
r
@Artun Duman how did you resolve this issue ?
a
I didn't πŸ™‚
p
Hi, going through the same exercise here I've started with deployment service spec, trying to pass the variables to the gateway using the cli but :
Copy code
spec:
      containers:
        - name: flink-sql-gateway
          image: flink:1.17.1-scala_2.12-java11
          command: ["/bin/sh"]
          args: ["-c", "/docker-entrypoint.sh /opt/flink/bin/sql-gateway.sh start-foreground -Dsql-gateway.endpoint.rest.address=0.0.0.0 -Djobmanager.rpc.address=flink-deployment -Dtaskmanager.host=flink-deployment -Drest.address=flink-deployment"]
          ports:
            - containerPort: 8083
gateway responds fine to GET 
/v1/info
but when trying to send a query via sql-client, it hangs + timeout and gateway logs I am seeing similar exception 
Connection refused: localhost/127.0.0.1:8081
I am guessing that the CLI 
-D
are not working (maybe default values from conf are being prioritized) or there is another property that needs to be set
it seems docker-entrypoint.sh is able to modify the conf file, will try pass via env or try write the conf file with the remote hostnames (not localhost)
Copy code
containers:
        - name: flink-sql-gateway
          image: flink:1.17.1-scala_2.12-java11
          env:
            - name: JOB_MANAGER_RPC_ADDRESS
              value: flink-deployment
          command: ["/bin/sh"]
          args: ["-c", "/docker-entrypoint.sh /opt/flink/bin/sql-gateway.sh start-foreground -Dsql-gateway.endpoint.rest.address=0.0.0.0"]
          ports:
            - containerPort: 8083
but not sure what 
rest.address
in the sql gateway should be, if ok 
0.0.0.0
or if this is another remote service
hm.. didn't really work for the sql-client to inherit that information what I found out that it worked was: after running:
Copy code
./bin/sql-client.sh gateway --endpoint <http://sql.flink.minikube>
I had to type
Copy code
SET rest.address=flink-deployment;
or running with the 
-Drest.address
:
Copy code
./bin/sql-client.sh gateway --endpoint <http://sql.flink.minikube> -Drest.address=flink-deployment
so that it would work. It seems the client is sending the params from its context, which is is then used by the gateway.
Copy code
Flink SQL> SELECT
>   name,
>   COUNT(*) AS cnt
> FROM
>   (VALUES ('Bob'), ('Alice'), ('Greg'), ('Bob')) AS NameTable(name)
> GROUP BY name;
+-------+-----+
|  name | cnt |
+-------+-----+
| Alice |   1 |
|   Bob |   2 |
|  Greg |   1 |
+-------+-----+
3 rows in set
r
@Pedro H S Teixeira When i run on k8s cluster( Not minickue) i am getting this error [ERROR] Could not execute SQL statement. Reason: org.apache.flink.runtime.rest.util.RestClientException: 413 Request Entity Too Large. Try to raise [rest.client.max-content-length]
i am running Flinksql > select 1;
Copy code
[ERROR] Could not execute SQL statement. Reason:
org.apache.flink.runtime.rest.util.RestClientException: 413 Request Entity Too Large. Try to raise [rest.client.max-content-length]
4 Views
Open in Slack
PreviousNext
Powered by