?????? flink on k8s native ????????

[????????????????]

Hi Yang??


????????????????????????????flink session on k8s ??examples ???????????????? 
application ??????????????????????????????


????????????????


????????????????


flink run-application \
--target kubernetes-application \
-Dkubernetes.namespace=flink-session-cluster \
-Dkubernetes.cluster-id=my-first-application-cluster \
-Dkubernetes.container.image=flink:latest \
local:///opt/flink/examples/batch/WordCount.jar



???? deployment ???? ??????????



????service???? ????????





????pod??????





???? describe pods ??????????





?????????????????????? configmap 
flink-config-my-first-application-cluster?????????????????? ?????? kubectl 
describe ?????????? flink-config-my-first-application-cluster????????????





?????? describe pods ??????????


Name:        
 my-first-application-cluster-6c85b474-64t6x
Namespace:    flink-session-cluster
Priority:     0
Node:         hkctttl104/10.199.252.104
Start Time:   Thu, 13 May 2021 22:16:03 +0800
Labels:       app=my-first-application-cluster
              component=jobmanager
              pod-template-hash=6c85b474
              type=flink-native-kubernetes
Annotations:&nbsp; <none&gt;
Status:&nbsp; &nbsp; &nbsp; &nbsp;Running
IP:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;10.47.0.8
IPs:
&nbsp; IP:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;10.47.0.8
Controlled By:&nbsp; ReplicaSet/my-first-application-cluster-6c85b474
Containers:
&nbsp; flink-job-manager:
&nbsp; &nbsp; Container ID:&nbsp; 
docker://0325efde4632560749515c225a89775414fb8e96dbd10757ede2b9e381c551f8
&nbsp; &nbsp; Image:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;flink:latest
&nbsp; &nbsp; Image ID:&nbsp; &nbsp; &nbsp; 
docker-pullable://flink@sha256:4911cd2b70f87a5286803fffb547c3676b7c3d078206547a4354be6749b1eef2
&nbsp; &nbsp; Ports:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp;8081/TCP, 6123/TCP, 
6124/TCP
&nbsp; &nbsp; Host Ports:&nbsp; &nbsp; 0/TCP, 0/TCP, 0/TCP
&nbsp; &nbsp; Command:
&nbsp; &nbsp; &nbsp; /docker-entrypoint.sh
&nbsp; &nbsp; Args:
&nbsp; &nbsp; &nbsp; native-k8s
&nbsp; &nbsp; &nbsp; $JAVA_HOME/bin/java -classpath $FLINK_CLASSPATH 
-Xmx1073741824 -Xms1073741824 -XX:MaxMetaspaceSize=268435456 
-Dlog.file=/opt/flink/log/jobmanager.log 
-Dlogback.configurationFile=file:/opt/flink/conf/logback-console.xml 
-Dlog4j.configuration=file:/opt/flink/conf/log4j-console.properties 
-Dlog4j.configurationFile=file:/opt/flink/conf/log4j-console.properties 
org.apache.flink.kubernetes.entrypoint.KubernetesApplicationClusterEntrypoint 
-D jobmanager.memory.off-heap.size=134217728b -D 
jobmanager.memory.jvm-overhead.min=201326592b -D 
jobmanager.memory.jvm-metaspace.size=268435456b -D 
jobmanager.memory.heap.size=1073741824b -D 
jobmanager.memory.jvm-overhead.max=201326592b
&nbsp; &nbsp; State:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Waiting
&nbsp; &nbsp; &nbsp; Reason:&nbsp; &nbsp; &nbsp; &nbsp;CrashLoopBackOff
&nbsp; &nbsp; Last State:&nbsp; &nbsp; &nbsp;Terminated
&nbsp; &nbsp; &nbsp; Reason:&nbsp; &nbsp; &nbsp; &nbsp;Error
&nbsp; &nbsp; &nbsp; Exit Code:&nbsp; &nbsp; 1
&nbsp; &nbsp; &nbsp; Started:&nbsp; &nbsp; &nbsp; Thu, 13 May 2021 22:27:59 
+0800
&nbsp; &nbsp; &nbsp; Finished:&nbsp; &nbsp; &nbsp;Thu, 13 May 2021 22:27:59 
+0800
&nbsp; &nbsp; Ready:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; False
&nbsp; &nbsp; Restart Count:&nbsp; 6
&nbsp; &nbsp; Limits:
&nbsp; &nbsp; &nbsp; cpu:&nbsp; &nbsp; &nbsp;1
&nbsp; &nbsp; &nbsp; memory:&nbsp; 1600Mi
&nbsp; &nbsp; Requests:
&nbsp; &nbsp; &nbsp; cpu:&nbsp; &nbsp; &nbsp;1
&nbsp; &nbsp; &nbsp; memory:&nbsp; 1600Mi
&nbsp; &nbsp; Environment:
&nbsp; &nbsp; &nbsp; _POD_IP_ADDRESS:&nbsp; &nbsp;(v1:status.podIP)
&nbsp; &nbsp; Mounts:
&nbsp; &nbsp; &nbsp; /opt/flink/conf from flink-config-volume (rw)
&nbsp; &nbsp; &nbsp; /var/run/secrets/kubernetes.io/serviceaccount from 
default-token-9qxvx (ro)
Conditions:
&nbsp; Type&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Status
&nbsp; Initialized&nbsp; &nbsp; &nbsp; &nbsp;True&nbsp;
&nbsp; Ready&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;False&nbsp;
&nbsp; ContainersReady&nbsp; &nbsp;False&nbsp;
&nbsp; PodScheduled&nbsp; &nbsp; &nbsp; True&nbsp;
Volumes:
&nbsp; flink-config-volume:
&nbsp; &nbsp; Type:&nbsp; &nbsp; &nbsp; ConfigMap (a volume populated by a 
ConfigMap)
&nbsp; &nbsp; Name:&nbsp; &nbsp; &nbsp; 
flink-config-my-first-application-cluster
&nbsp; &nbsp; Optional:&nbsp; false
&nbsp; default-token-9qxvx:
&nbsp; &nbsp; Type:&nbsp; &nbsp; &nbsp; &nbsp; Secret (a volume populated by a 
Secret)
&nbsp; &nbsp; SecretName:&nbsp; default-token-9qxvx
&nbsp; &nbsp; Optional:&nbsp; &nbsp; false
QoS Class:&nbsp; &nbsp; &nbsp; &nbsp;Guaranteed
Node-Selectors:&nbsp; <none&gt;
Tolerations:&nbsp; &nbsp; &nbsp;node.kubernetes.io/not-ready:NoExecute for 300s
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp;node.kubernetes.io/unreachable:NoExecute for 300s
Events:
&nbsp; Type&nbsp; &nbsp; &nbsp;Reason&nbsp; &nbsp; &nbsp; &nbsp;Age&nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;From&nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;Message
&nbsp; ----&nbsp; &nbsp; &nbsp;------&nbsp; &nbsp; &nbsp; &nbsp;----&nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; ----&nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;-------
&nbsp; Normal&nbsp; &nbsp;Scheduled&nbsp; &nbsp; 14m&nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;default-scheduler&nbsp; 
Successfully assigned 
flink-session-cluster/my-first-application-cluster-6c85b474-64t6x to hkctttl104
&nbsp; Warning&nbsp; FailedMount&nbsp; 14m&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;kubelet&nbsp; &nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; MountVolume.SetUp failed for volume "flink-config-volume" : 
configmap "flink-config-my-first-application-cluster" not found
&nbsp; Normal&nbsp; &nbsp;Pulling&nbsp; &nbsp; &nbsp; 14m&nbsp; &nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;kubelet&nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; Pulling image "flink:latest"
&nbsp; Normal&nbsp; &nbsp;Pulled&nbsp; &nbsp; &nbsp; &nbsp;8m46s&nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;kubelet&nbsp; &nbsp; 
&nbsp; &nbsp; &nbsp; &nbsp; Successfully pulled image "flink:latest"
&nbsp; Normal&nbsp; &nbsp;Created&nbsp; &nbsp; &nbsp; 7m8s (x5 over 
8m46s)&nbsp; &nbsp; kubelet&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Created 
container flink-job-manager
&nbsp; Normal&nbsp; &nbsp;Started&nbsp; &nbsp; &nbsp; 7m8s (x5 over 
8m45s)&nbsp; &nbsp; kubelet&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Started 
container flink-job-manager
&nbsp; Normal&nbsp; &nbsp;Pulled&nbsp; &nbsp; &nbsp; &nbsp;7m8s (x4 over 
8m45s)&nbsp; &nbsp; kubelet&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Container 
image "flink:latest" already present on machine
&nbsp; Warning&nbsp; BackOff&nbsp; &nbsp; &nbsp; 4m36s (x23 over 8m43s)&nbsp; 
kubelet&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; Back-off restarting failed 
container





??????????????????????????????


Best??


Luke Yan




------------------&nbsp;????????&nbsp;------------------
??????:                                                                         
                                               "user-zh"                        
                                                            
<danrtsey...@gmail.com&gt;;
????????:&nbsp;2021??5??12??(??????) ????5:42
??????:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;;

????:&nbsp;Re: flink on k8s native ????????



??????????????????????kubernetes.rest-service.exposed.type=NodePort

Best,
Yang

???????????????? <1510603...@qq.com&gt; ??2021??5??12?????? ????9:45??????

&gt; Hi All??
&gt;
&gt; ?????????????????? k8s ???????????????????? flink on k8s native
&gt; ??????????????????????????????????????????????????????????????????????
&gt;
&gt; ????????
&gt;
&gt;&nbsp;&nbsp;&nbsp;&nbsp; k8s&nbsp; v1.18.16
&gt;&nbsp;&nbsp;&nbsp;&nbsp; flink&nbsp; flink-1.12.2
&gt;&nbsp;&nbsp;&nbsp;&nbsp; java&nbsp; openjdk version "11" 2018-09-25
&gt;
&gt; ??????????????????????
&gt;
&gt;&nbsp;&nbsp;&nbsp; 1??Kubernetes &gt;= 1.9&nbsp; ????
&gt;
&gt;
&gt;
&gt;&nbsp;&nbsp;&nbsp;&nbsp; 2??????????&nbsp; ????
&gt;
&gt;
&gt;&nbsp; 3?????? Kubernetes DNS&nbsp; ????
&gt;
&gt;
&gt; 4??default service account with RBAC
&gt; 
<https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/deployment/resource-providers/native_kubernetes/#rbac&gt;
 permissions
&gt; to create, delete pods ???????????????? 
??????????????????????????????????????
&gt;
&gt; kubectl create namespace flink-session-cluster
&gt;
&gt; kubectl create serviceaccount flink -n flink-session-cluster
&gt;
&gt; kubectl create clusterrolebinding flink-role-binding-flink \ --clusterrole
&gt; =edit \ --serviceaccount=flink-session-cluster:flink
&gt;
&gt;
&gt; ???????? session ?????? flink ????
&gt;
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; 3.1 ????session ????
&gt;
&gt; ./bin/kubernetes-session.sh -Dkubernetes.cluster-id=my-first-flink-cluster
&gt; \
&gt; &gt; -Dkubernetes.namespace=flink-session-cluster \
&gt; &gt; -Dkubernetes.jobmanager.service-account=flink \
&gt; &gt; -Dkubernetes.service.exposed.type=NodePort
&gt;
&gt;&nbsp;&nbsp; ??????
&gt;
&gt;
&gt;&nbsp;&nbsp; 3.2 ????flink ????
&gt;
&gt;&nbsp;&nbsp;&nbsp; ??????
&gt;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; ./bin/flink run \
&gt; &gt;&nbsp;&nbsp;&nbsp;&nbsp; --target kubernetes-session \
&gt; &gt; -Dkubernetes.namespace=flink-session-cluster \
&gt; &gt;&nbsp;&nbsp;&nbsp;&nbsp; 
-Dkubernetes.cluster-id=my-first-flink-cluster \
&gt; &gt; -Dkubernetes.jobmanager.service-account=flink \
&gt; &gt;&nbsp;&nbsp;&nbsp;&nbsp; ./examples/streaming/TopSpeedWindowing.jar
&gt;
&gt;&nbsp;&nbsp;&nbsp; ??????????????
&gt;
&gt; Executing TopSpeedWindowing example with default input data set.
&gt; Use --input to specify file input.
&gt; Printing result to stdout. Use --output to specify output path.
&gt; WARNING: An illegal reflective access operation has occurred
&gt; WARNING: Illegal reflective access by
&gt; org.apache.flink.api.java.ClosureCleaner
&gt; (file:/home/dsi/soft/flink/flink-1.12.2/lib/flink-dist_2.11-1.12.2.jar) to
&gt; field java.lang.String.value
&gt; WARNING: Please consider reporting this to the maintainers of
&gt; org.apache.flink.api.java.ClosureCleaner
&gt; WARNING: Use --illegal-access=warn to enable warnings of further illegal
&gt; reflective access operations
&gt; WARNING: All illegal access operations will be denied in a future release
&gt; 2021-05-12 15:51:30,453 INFO
&gt; 
org.apache.flink.kubernetes.KubernetesClusterDescriptor&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 [] - Retrieve
&gt; flink cluster my-first-flink-cluster successfully, JobManager Web
&gt; Interface: http://10.199.252.101:8081
&gt;
&gt; ------------------------------------------------------------
&gt;&nbsp; The program finished with the following exception:
&gt;
&gt; org.apache.flink.client.program.ProgramInvocationException: The main
&gt; method caused an error: Failed to execute job 
'CarTopSpeedWindowingExample'.
&gt; at
&gt; 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:366)
&gt; at
&gt; 
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:219)
&gt; at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
&gt; at
&gt; 
org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
&gt; at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
&gt; at
&gt; org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
&gt; at
&gt; 
org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
&gt; at
&gt; 
org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
&gt; at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
&gt; Caused by: org.apache.flink.util.FlinkException: Failed to execute job
&gt; 'CarTopSpeedWindowingExample'.
&gt; at
&gt; 
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.executeAsync(StreamExecutionEnvironment.java:1918)
&gt; at
&gt; 
org.apache.flink.client.program.StreamContextEnvironment.executeAsync(StreamContextEnvironment.java:135)
&gt; at
&gt; 
org.apache.flink.client.program.StreamContextEnvironment.execute(StreamContextEnvironment.java:76)
&gt; at
&gt; 
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.execute(StreamExecutionEnvironment.java:1782)
&gt; at
&gt; 
org.apache.flink.streaming.examples.windowing.TopSpeedWindowing.main(TopSpeedWindowing.java:99)
&gt; at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native
&gt; Method)
&gt; at
&gt; 
java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
&gt; at
&gt; 
java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
&gt; at java.base/java.lang.reflect.Method.invoke(Method.java:566)
&gt; at
&gt; 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:349)
&gt; ... 8 more
&gt; Caused by: org.apache.flink.runtime.client.JobSubmissionException: Failed
&gt; to submit JobGraph.
&gt; at
&gt; 
org.apache.flink.client.program.rest.RestClusterClient.lambda$submitJob$7(RestClusterClient.java:400)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture.uniExceptionally(CompletableFuture.java:986)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture$UniExceptionally.tryFire(CompletableFuture.java:970)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
&gt; at
&gt; 
org.apache.flink.runtime.concurrent.FutureUtils.lambda$retryOperationWithDelay$9(FutureUtils.java:390)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture.completeExceptionally(CompletableFuture.java:2088)
&gt; at
&gt; 
org.apache.flink.runtime.rest.RestClient.lambda$submitRequest$1(RestClient.java:430)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:577)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:570)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:549)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:490)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.setValue0(DefaultPromise.java:615)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.setFailure0(DefaultPromise.java:608)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:117)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:321)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:337)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:702)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
&gt; at java.base/java.lang.Thread.run(Thread.java:834)
&gt; Caused by: org.apache.flink.runtime.concurrent.FutureUtils$RetryException:
&gt; Could not complete the operation. Number of retries has been exhausted.
&gt; at
&gt; 
org.apache.flink.runtime.concurrent.FutureUtils.lambda$retryOperationWithDelay$9(FutureUtils.java:386)
&gt; ... 21 more
&gt; Caused by: java.util.concurrent.CompletionException:
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AnnotatedConnectException:
&gt; Connection refused: /10.199.252.101:8081
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:331)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:346)
&gt; at
&gt; 
java.base/java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:1063)
&gt; ... 19 more
&gt; Caused by:
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AnnotatedConnectException:
&gt; Connection refused: /10.199.252.101:8081
&gt; Caused by: java.net.ConnectException: Connection refused
&gt; at java.base/sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)
&gt; at
&gt; 
java.base/sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:779)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:330)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:334)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:702)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:650)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:576)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:493)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:989)
&gt; at
&gt; 
org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
&gt; at java.base/java.lang.Thread.run(Thread.java:834)
&gt;
&gt;
&gt; Best??
&gt;
&gt; Luke Yan
&gt;
&gt;