https://flink.apache.org/ logo
Join Slack
Powered by
Hi, I’m trying to deploy a `SessionJob` using `fl...
# troubleshooting
h
Hi, I’m trying to deploy a 
SessionJob
using 
flink-kubernetes-operator
. However I’m receiving this event on the `SessionJob`: 
"Connection timed out (Connection timed out)"
Looking in the operator logs I see this:
Copy code
flink-kubernetes-operator 2023-04-17 21:40:34,212 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] Starting reconciliation
flink-kubernetes-operator 2023-04-17 21:40:34,213 o.a.f.k.o.s.FlinkResourceContextFactory [INFO ][data-team/flink-cluster] Getting service for flink-cluster
flink-kubernetes-operator 2023-04-17 21:40:34,223 o.a.f.k.o.r.d.AbstractFlinkResourceReconciler [INFO ][data-team/flink-cluster] Resource fully reconciled, nothing to do...
flink-kubernetes-operator 2023-04-17 21:40:34,223 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] End of reconciliation
flink-kubernetes-operator 2023-04-17 21:40:49,224 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] Starting reconciliation
flink-kubernetes-operator 2023-04-17 21:40:49,225 o.a.f.k.o.s.FlinkResourceContextFactory [INFO ][data-team/flink-cluster] Getting service for flink-cluster
flink-kubernetes-operator 2023-04-17 21:40:49,233 o.a.f.k.o.r.d.AbstractFlinkResourceReconciler [INFO ][data-team/flink-cluster] Resource fully reconciled, nothing to do...
flink-kubernetes-operator 2023-04-17 21:40:49,233 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] End of reconciliation
flink-kubernetes-operator 2023-04-17 21:41:04,234 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] Starting reconciliation
flink-kubernetes-operator 2023-04-17 21:41:04,235 o.a.f.k.o.s.FlinkResourceContextFactory [INFO ][data-team/flink-cluster] Getting service for flink-cluster
flink-kubernetes-operator 2023-04-17 21:41:04,244 o.a.f.k.o.r.d.AbstractFlinkResourceReconciler [INFO ][data-team/flink-cluster] Resource fully reconciled, nothing to do...
flink-kubernetes-operator 2023-04-17 21:41:04,244 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] End of reconciliation
flink-kubernetes-operator 2023-04-17 21:41:19,246 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] Starting reconciliation
flink-kubernetes-operator 2023-04-17 21:41:19,246 o.a.f.k.o.s.FlinkResourceContextFactory [INFO ][data-team/flink-cluster] Getting service for flink-cluster
flink-kubernetes-operator 2023-04-17 21:41:19,254 o.a.f.k.o.r.d.AbstractFlinkResourceReconciler [INFO ][data-team/flink-cluster] Resource fully reconciled, nothing to do...
flink-kubernetes-operator 2023-04-17 21:41:19,254 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] End of reconciliation
flink-kubernetes-operator 2023-04-17 21:41:30,342 o.a.f.k.o.l.AuditUtils         [INFO ][data-team/flink-job] >>> Event  | Warning | SESSIONJOBEXCEPTION | Connection timed out (Connection timed out)
flink-kubernetes-operator 2023-04-17 21:41:30,342 o.a.f.k.o.r.ReconciliationUtils [WARN ][data-team/flink-job] Attempt count: 3, last attempt: false
flink-kubernetes-operator 2023-04-17 21:41:30,356 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] Starting reconciliation
flink-kubernetes-operator 2023-04-17 21:41:30,356 o.a.f.k.o.s.FlinkResourceContextFactory [INFO ][data-team/flink-cluster] Getting service for flink-cluster
flink-kubernetes-operator 2023-04-17 21:41:30,364 o.a.f.k.o.r.d.AbstractFlinkResourceReconciler [INFO ][data-team/flink-cluster] Resource fully reconciled, nothing to do...
flink-kubernetes-operator 2023-04-17 21:41:30,364 o.a.f.k.o.c.FlinkDeploymentController [INFO ][data-team/flink-cluster] End of reconciliation
flink-kubernetes-operator 2023-04-17 21:41:30,375 o.a.f.k.o.l.AuditUtils         [INFO ][data-team/flink-job] >>> Status | Error   | UPGRADING       | {"type":"org.apache.flink.kubernetes.operator.exception.ReconciliationException","message":"java.net.ConnectException: Connection timed out (Connection timed out)","throwableList":[{"type":"java.net.ConnectException","message":"Connection timed out (Connection timed out)"}]} 
flink-kubernetes-operator 2023-04-17 21:41:30,375 i.j.o.p.e.ReconciliationDispatcher [ERROR][data-team/flink-job] Error during event processing ExecutionScope{ resource id: ResourceID{name='flink-job', namespace='data-team'}, version: 1307816475} failed.
flink-kubernetes-operator org.apache.flink.kubernetes.operator.exception.ReconciliationException: java.net.ConnectException: Connection timed out (Connection timed out)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController.reconcile(FlinkSessionJobController.java:112)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController.reconcile(FlinkSessionJobController.java:52)
flink-kubernetes-operator     at io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:145)
flink-kubernetes-operator     at io.javaoperatorsdk.operator.processing.Controller$1.execute(Controller.java:103)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.metrics.OperatorJosdkMetrics.timeControllerExecution(OperatorJosdkMetrics.java:80)
flink-kubernetes-operator     at io.javaoperatorsdk.operator.processing.Controller.reconcile(Controller.java:102)
flink-kubernetes-operator     at io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.reconcileExecution(ReconciliationDispatcher.java:139)
flink-kubernetes-operator     at io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleReconcile(ReconciliationDispatcher.java:119)
flink-kubernetes-operator     at io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleDispatch(ReconciliationDispatcher.java:89)
flink-kubernetes-operator     at io.javaoperatorsdk.operator.processing.event.ReconciliationDispatcher.handleExecution(ReconciliationDispatcher.java:62)
flink-kubernetes-operator     at io.javaoperatorsdk.operator.processing.event.EventProcessor$ReconcilerExecutor.run(EventProcessor.java:406)
flink-kubernetes-operator     at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
flink-kubernetes-operator     at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
flink-kubernetes-operator     at java.base/java.lang.Thread.run(Unknown Source)
flink-kubernetes-operator Caused by: java.net.ConnectException: Connection timed out (Connection timed out)
flink-kubernetes-operator     at java.base/java.net.PlainSocketImpl.socketConnect(Native Method)
flink-kubernetes-operator     at java.base/java.net.AbstractPlainSocketImpl.doConnect(Unknown Source)
flink-kubernetes-operator     at java.base/java.net.AbstractPlainSocketImpl.connectToAddress(Unknown Source)
flink-kubernetes-operator     at java.base/java.net.AbstractPlainSocketImpl.connect(Unknown Source)
flink-kubernetes-operator     at java.base/java.net.SocksSocketImpl.connect(Unknown Source)
flink-kubernetes-operator     at java.base/java.net.Socket.connect(Unknown Source)
flink-kubernetes-operator     at java.base/sun.security.ssl.SSLSocketImpl.connect(Unknown Source)
flink-kubernetes-operator     at java.base/sun.security.ssl.BaseSSLSocketImpl.connect(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.NetworkClient.doConnect(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.http.HttpClient.openServer(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.http.HttpClient.openServer(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.https.HttpsClient.<init>(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.https.HttpsClient.New(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.https.AbstractDelegateHttpsURLConnection.getNewHttpClient(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.http.HttpURLConnection.plainConnect0(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.http.HttpURLConnection.plainConnect(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.https.AbstractDelegateHttpsURLConnection.connect(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream0(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream(Unknown Source)
flink-kubernetes-operator     at java.base/sun.net.www.protocol.https.HttpsURLConnectionImpl.getInputStream(Unknown Source)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.artifact.HttpArtifactFetcher.fetch(HttpArtifactFetcher.java:59)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.artifact.ArtifactManager.fetch(ArtifactManager.java:61)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.service.AbstractFlinkService.uploadJar(AbstractFlinkService.java:724)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.service.AbstractFlinkService.submitJobToSessionCluster(AbstractFlinkService.java:207)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.reconciler.sessionjob.SessionJobReconciler.deploy(SessionJobReconciler.java:70)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.reconciler.sessionjob.SessionJobReconciler.deploy(SessionJobReconciler.java:41)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.reconciler.deployment.AbstractFlinkResourceReconciler.reconcile(AbstractFlinkResourceReconciler.java:115)
flink-kubernetes-operator     at org.apache.flink.kubernetes.operator.controller.FlinkSessionJobController.reconcile(FlinkSessionJobController.java:104)
flink-kubernetes-operator     ... 13 more
Could anyone help me debug that?
Here is the SessionJob manifest:
Copy code
apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkSessionJob
metadata:
  annotations:
    <http://kubectl.kubernetes.io/last-applied-configuration|kubectl.kubernetes.io/last-applied-configuration>: >
      {"apiVersion":"<http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>","kind":"FlinkSessionJob","metadata":{"annotations":{},"labels":{"<http://argocd.argoproj.io/instance|argocd.argoproj.io/instance>":"dev-ga-flink-jobs"},"name":"flink-job","namespace":"data-team"},"spec":{"deploymentName":"flink-cluster","flinkConfiguration":{},"job":{"jarURI":"<https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.16.1/flink-examples-streaming_2.12-1.16.1-WordCount.jar>","parallelism":1,"state":"running","upgradeMode":"stateless"}}}
  creationTimestamp: '2023-04-17T21:32:29Z'
  finalizers:
    - <http://flinksessionjobs.flink.apache.org/finalizer|flinksessionjobs.flink.apache.org/finalizer>
  generation: 1
  labels:
    <http://argocd.argoproj.io/instance|argocd.argoproj.io/instance>: dev-ga-flink-jobs
    target.session: flink-cluster
  name: flink-job
  namespace: data-team
  resourceVersion: '1307824167'
  uid: d660649d-1f58-4eb3-b257-82d765ec596e
spec:
  deploymentName: flink-cluster
  flinkConfiguration: {}
  job:
    args: []
    jarURI: >-
      <https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.16.1/flink-examples-streaming_2.12-1.16.1-WordCount.jar>
    parallelism: 1
    state: running
    upgradeMode: stateless
status:
  error: >-
    {"type":"org.apache.flink.kubernetes.operator.exception.ReconciliationException","message":"java.net.ConnectException:
    Connection timed out (Connection timed
    out)","throwableList":[{"type":"java.net.ConnectException","message":"Connection
    timed out (Connection timed out)"}]}
  jobStatus:
    savepointInfo:
      lastPeriodicSavepointTimestamp: 0
      savepointHistory: []
  lifecycleState: UPGRADING
  reconciliationStatus:
    lastReconciledSpec: >-
      {"spec":{"job":{"jarURI":"<https://repo1.maven.org/maven2/org/apache/flink/flink-examples-streaming_2.12/1.16.1/flink-examples-streaming_2.12-1.16.1-WordCount.jar>","parallelism":1,"entryClass":null,"args":[],"state":"suspended","savepointTriggerNonce":null,"initialSavepointPath":null,"upgradeMode":"stateless","allowNonRestoredState":null},"restartNonce":null,"flinkConfiguration":{},"deploymentName":"flink-cluster"},"resource_metadata":{"apiVersion":"<http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>","metadata":{"generation":1},"firstDeployment":true}}
    reconciliationTimestamp: 1681767710417
    state: UPGRADING
The job works when deployed in Application Mode but not in Session Mode
t
1. Verify the URL of the artifact (JAR file) that the operator is trying to fetch. Ensure that the URL is correct and accessible from the Kubernetes cluster where the operator is running. 2. Check the network connectivity between the operator and the HTTP endpoint hosting the artifact. This may involve checking firewall rules, network policies, and other network configurations. 3. If the issue persists, you may want to increase the connection timeout for the HTTP client used by the operator. This can be done by modifying the operator's configuration or code, depending on how the operator is implemented.
h
Oh, I see that this endpoint is not accessible from within the Operator pods. Makes sense
Thank you
Why does it work when deployed as an Application Cluster?
Ok. Now I’m having issues when trying to run a JAR that is stored in S3. I have created a custom 
flink-kubernetes-operator
image with the plugins. But I’m getting this issue:
Copy code
flink-kubernetes-operator Caused by: org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException: No AWS Credentials provided by TemporaryAWSCredentialsProvider SimpleAWSCredentialsProvider EnvironmentVariableCredentialsProvider IAMInstanceCredentialsProvider : com.amazonaws.SdkClientException: Unable to load AWS credentials from environment variables (AWS_ACCESS_KEY_ID (or AWS_ACCESS_KEY) and AWS_SECRET_KEY (or AWS_SECRET_ACCESS_KEY))
On the deployment side I have fixed this by setting 
fs.s3a.aws.credentials.provider: com.amazonaws.auth.WebIdentityTokenCredentialsProvider
. But on the operator side it seems like I don’t have this option
t
The error message also mentions that the 
EnvironmentVariableCredentialsProvider
was unable to load AWS credentials from environment variables (
AWS_ACCESS_KEY_ID
and 
AWS_SECRET_ACCESS_KEY
). To resolve this issue, you need to provide valid AWS credentials to the Flink Kubernetes operator. There are several ways to do this: 1. Environment Variables: Set the 
AWS_ACCESS_KEY_ID
and 
AWS_SECRET_ACCESS_KEY
environment variables with your AWS access key and secret key, respectively. You can do this by modifying the Kubernetes deployment or pod specification for the Flink Kubernetes operator to include these environment variables. 2. IAM Role: If the Flink Kubernetes operator is running on an Amazon EC2 instance or an Amazon EKS cluster, you can assign an IAM role to the instance or the Kubernetes worker nodes. This IAM role should have the necessary permissions to access the S3 bucket. The 
IAMInstanceCredentialsProvider
will automatically use the IAM role associated with the instance. After providing the credentials using one of the methods above, restart the Flink Kubernetes operator and verify that it can successfully access the S3 bucket.
h
The Flink Operator is running on EKS and its service account already has an IAM role
Is it possible to set the operator to use the 
WebIdentityTokenCredentialsProvider
?
t
m
@Taruj Goyal @Hygor Knust Hello,
Copy code
fs.s3.aws.credentials.provider: com.amazonaws.auth.WebIdentityTokenCredentialsProvider
    fs.s3.path.style.access: 'true'
    fs.s3.impl: org.apache.hadoop.fs.s3a.S3AFileSystem
    fs.s3.endpoint: s3.<my_region>.<http://amazonaws.com|amazonaws.com>
I set configuration like this and the errors disappeared. But it is stucking at this part; 
2023-04-19 06:50:13,086 INFO  org.apache.flink.runtime.blob.FileSystemBlobStore            [] - Creating highly available BLOB storage directory at s3://<my_bucket>/<my_path>/blob
h
Yes, I have this configuration, in the Flink Deployment using the 
flinkConfiguration
in the Helm Chart and it is working. My issue is to set this on the Flink Operator, it seems like there is no value in the Helm Chart to set it https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/operations/helm/. I have tried setting this on the 
flink-kubernetes-operator
Helm deployment:
Copy code
defaultConfiguration:
  create: true
  append: true
  flink-conf.yaml: |+
    fs.s3a.aws.credentials.provider: com.amazonaws.auth.WebIdentityTokenCredentialsProvider
But it seems like the Flink Operator itself does not use this configuration.
I have this env in the 
flink-kubernetes-operator
pod:
Copy code
AWS_STS_REGIONAL_ENDPOINTS:   regional
AWS_DEFAULT_REGION:           us-west-2
AWS_REGION:                   us-west-2
AWS_ROLE_ARN:                 arn:aws:iam::{account}:role/{role_name}
AWS_WEB_IDENTITY_TOKEN_FILE:  /var/run/secrets/eks.amazonaws.com/serviceaccount/token
Copy code
jvmArgs:
    operator: "-Daws.webIdentityTokenFile=/var/run/secrets/eks.amazonaws.com/serviceaccount/token"
Also set this in the Helm values but no luck.
23 Views
Open in Slack
PreviousNext
Powered by