<p>I'm using the Apache Flink Kubernetes operator to deploy a standalone job on an Application cluster setup.</p>
<p>I have setup the following files using the Flink official documentation - <a href="https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/deployment/resource-providers/standalone/kubernetes/" rel="nofollow noreferrer">Link</a></p>
<ol>
<li>jobmanager-application-non-ha.yaml</li>
<li>taskmanager-job-deployment.yaml</li>
<li>flink-configuration-configmap.yaml</li>
<li>jobmanager-service.yaml</li>
</ol>
<p>I have not changed any of the configurations in these files and am trying to run a simple WordCount example from the Flink examples using the Apache Flink Operator.</p>
<p>After running the kubectl commands to setting up the job manager and the task manager - the job manager goes into a NotReady state while the task manager goes into a CrashLoopBackOff loop.</p>
<pre><code>NAME                                         READY   STATUS             RESTARTS        AGE
flink-jobmanager-28k4b                       1/2     NotReady           2 (4m24s ago)   16m
flink-kubernetes-operator-6585dddd97-9hjp4   2/2     Running            0               10d
flink-taskmanager-6bb88468d7-ggx8t           1/2     CrashLoopBackOff   9 (2m21s ago)   15m
</code></pre>
<p>The job manager logs look like this</p>
<pre><code>org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException: Slot request bulk is not fulfillable! Could not allocate the required slot within slot request timeout
    at org.apache.flink.runtime.jobmaster.slotpool.PhysicalSlotRequestBulkCheckerImpl.lambda$schedulePendingRequestBulkWithTimestampCheck$0(PhysicalSlotRequestBulkCheckerImpl.java:86) ~[flink-dist-1.16.0.jar:1.16.0]
    at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source) ~[?:?]
    at java.util.concurrent.FutureTask.run(Unknown Source) ~[?:?]
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRunAsync$4(AkkaRpcActor.java:453) ~[flink-rpc-akka_be40712e-8b2e-47cd-baaf-f0149cf2604d.jar:1.16.0]
    at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68) ~[flink-rpc-akka_be40712e-8b2e-47cd-baaf-f0149cf2604d.jar:1.16.0]
</code></pre>
<p>The Task manager it seems cannot connect to the job manager</p>
<pre><code>2023-01-28 19:21:47,647 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Connecting to ResourceManager akka.tcp://flink@flink-jobmanager:6123/user/rpc/resourcemanager_*(00000000000000000000000000000000).
2023-01-28 19:21:57,766 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Could not resolve ResourceManager address akka.tcp://flink@flink-jobmanager:6123/user/rpc/resourcemanager_*, retrying in 10000 ms: Could not connect to rpc endpoint under address akka.tcp://flink@flink-jobmanager:6123/user/rpc/resourcemanager_*.
2023-01-28 19:22:08,036 INFO  akka.remote.transport.ProtocolStateActor                     [] - No response from remote for outbound association. Associate timed out after [20000 ms].
2023-01-28 19:22:08,057 WARN  akka.remote.ReliableDeliverySupervisor                       [] - Association with remote system [akka.tcp://flink@flink-jobmanager:6123] has failed, address is now gated for [50] ms. Reason: [Association failed with [akka.tcp://flink@flink-jobmanager:6123]] Caused by: [No response from remote for outbound association. Associate timed out after [20000 ms].]
2023-01-28 19:22:08,069 INFO  org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Could not resolve ResourceManager address akka.tcp://flink@flink-jobmanager:6123/user/rpc/resourcemanager_*, retrying in 10000 ms: Could not connect to rpc endpoint under address akka.tcp://flink@flink-jobmanager:6123/user/rpc/resourcemanager_*.
2023-01-28 19:22:08,308 WARN  akka.remote.transport.netty.NettyTransport                   [] - Remote connection to [null] failed with org.jboss.netty.channel.ConnectTimeoutException: connection timed out: flink-jobmanager/100.127.18.9:6123
</code></pre>
<p>The flink-configuration-configmap.yaml looks like this</p>
<pre><code>  flink-conf.yaml: |+
    jobmanager.rpc.address: flink-jobmanager
    taskmanager.numberOfTaskSlots: 2
    blob.server.port: 6124
    jobmanager.rpc.port: 6123
    taskmanager.rpc.port: 6122
    queryable-state.proxy.ports: 6125
    jobmanager.memory.process.size: 1600m
    taskmanager.memory.process.size: 1728m
    parallelism.default: 2    
</code></pre>
<p>This is what the pom.xml looks like - <a href="https://github.com/shroffrushabh/WordCount/blob/main/pom.xml" rel="nofollow noreferrer">Link</a></p>

user1386101
<p>You deployed the Kubernetes Operator in the namespace, but you did not create the CRDs the Operator requires. Instead you tried to create a standalone Flink Kubernetes cluster.</p>
<p>The Flink Operator makes it a lot easier to deploy your Flink jobs, you only need to deploy the operator itself and <code>FlinkDeployment</code>/<code>FlinkSessionJob</code> CRDs. The operator will manage your deployment after.</p>
<p>Please use this documentation for the Kubernetes Operator: <a href="https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/try-flink-kubernetes-operator/quick-start/" rel="nofollow noreferrer">Link</a></p>
