https://flink.apache.org/ logo
Join Slack
Powered by
Hello, Flink supports <pluggable file systems> an...
# troubleshooting
j
Hello, Flink supports pluggable file systems and, to use a specific file system eg) S3, the relevant jar file should exist under /opt/flink/plugins. I have two questions about reading/writing files in S3. 1. Is it possible to specify a plugin path when I run a flink application locally (MiniCluster)? 2. Can the file system jar file (eg flink-s3-fs-hadoop-flink-verion.jar) be added to a single Fat Jar file? I followed an example from aws-samples but I had the following error shown in the screen shot.
At least, I'm able to run the app without an error in a Flink cluster deployed on Docker. Dockerfile
Copy code
FROM flink:1.15.2

ARG PYTHON_VERSION
ENV PYTHON_VERSION=${PYTHON_VERSION:-3.8.10}
ARG FLINK_VERSION
ENV FLINK_VERSION=${FLINK_VERSION:-1.15.2}

RUN mkdir ./plugins/s3-fs-hadoop \
  && cp ./opt/flink-s3-fs-hadoop-${FLINK_VERSION}.jar ./plugins/s3-fs-hadoop 

... install python and pyflink
Ideally it'll be good if the plugin is pre-installed in Amazon Managed Flink. I need to check...
So far it is okay to run on a local Flink cluster deployed on Docker. I built a custom fat jar for connecting to a Kafka cluster on Amazon MSK as it is required by Amazon managed Flink, and I excluded the flink-s3-fs-hadoop-1.15.2.jar file when building it. When I located the S3 jar file under the plugins folder in the docker container, it worked without an issue. Therefore, I speculated the S3 jar file would also exist in the managed Flink environment and deployed the Flink app tp it as it is. However it failed and its error code indicates that the application code is invalid. Even it failed with the same error when I updated the fat jar file based on an application from aws-samples. I created a support ticket and, hopefully, it would be resolved with help of AWS support.
Copy code
{
  "applicationARN": "arn:aws:kinesisanalytics:ap-southeast-2:590312749310:application/real-time-streaming-loader-app",
  "applicationVersionId": 1,
  "message": "org.apache.flink.runtime.rest.handler.RestHandlerException: Could not execute application.\n\tat org.apache.flink.runtime.webmonitor.handlers.JarRunOverrideHandler.lambda$handleRequest$4(JarRunOverrideHandler.java:248)\n\tat java.base/java.util.concurrent.CompletableFuture.uniHandle(CompletableFuture.java:930)\n\tat java.base/java.util.concurrent.CompletableFuture$UniHandle.tryFire(CompletableFuture.java:907)\n\tat java.base/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)\n\tat java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1705)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\nCaused by: java.util.concurrent.CompletionException: org.apache.flink.client.program.ProgramAbortException: java.lang.RuntimeException: Python process exits with code: 1\n\tat java.base/java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:314)\n\tat java.base/java.util.concurrent.CompletableFuture.completeThrowable(CompletableFuture.java:319)\n\tat java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1702)\n\t... 6 more\nCaused by: org.apache.flink.client.program.ProgramAbortException: java.lang.RuntimeException: Python process exits with code: 1\n\tat org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:140)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)\n\tat java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\n\tat java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\n\tat java.base/java.lang.reflect.Method.invoke(Method.java:566)\n\tat org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)\n\tat org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)\n\tat org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)\n\tat org.apache.flink.client.deployment.application.DetachedApplicationRunner.tryExecuteJobs(DetachedApplicationRunner.java:84)\n\tat org.apache.flink.client.deployment.application.DetachedApplicationRunner.run(DetachedApplicationRunner.java:70)\n\tat org.apache.flink.runtime.webmonitor.handlers.JarRunOverrideHandler.lambda$handleRequest$3(JarRunOverrideHandler.java:239)\n\tat java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)\n\t... 6 more\nCaused by: java.lang.RuntimeException: Python process exits with code: 1\n\tat org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:130)\n\t... 17 more\n",
  "messageType": "ERROR",
  "messageSchemaVersion": "1",
  "errorCode": "CodeError.InvalidApplicationCode"
}
Open in Slack
PreviousNext
Powered by