Hello everyone 🙂 I have a question regarding Flink k8s operator, maybe someone knows a way around it. I am using persistent volume for operator, which is used to upload JARs. I created standalone

FlinkDeployment

. Once those JARs are used by

FlinkSessionJob

, they get duplicated with random hash in front of the name, which makes “Submit New Job” widget in GUI hard to navigate through. Is there an option for operator so that once I create

FlinkSessionJob

, that it does not create stand-alone JAR for each instance of job that is using this JAR? Example of `FlinkSessionJob`:

apiVersion: <http://flink.apache.org/v1beta1|flink.apache.org/v1beta1>
kind: FlinkSessionJob
metadata:
  name: test
  namespace: flink
spec:
  deploymentName: flink-test
  job:
    jarURI: file:///data/uploads/flink-web-upload/84c65347-e334-4811-869f-8f4f0b85d698_flink-test-ingest-0.1.8-SNAPSHOT.jar
    parallelism: 1
    entryClass: at.test.flink.ingest.SimpleTestSqlForwardingJob
    args:
    - "-dataType"
    - "test-10min"
    - "-kafkaTopic"
    - "test-10min"
    - "-eventhubProperties"
    - "/opt/flink/conf-extra/test-fanout.config"
    - "-enableJsonValidation"
    - "true"
    - "-offset"
    - "EARLIEST"
    - "-kafkaGroupId"
    - "test-10min-flink-consumer"
    state: running
    upgradeMode: savepoint

Once I upload JAR to Flink, I have only

flink-test-ingest-0.1.8-SNAPSHOT.jar

in GUI, but once I start the job, it also adds

84c65347-e334-4811-869f-8f4f0b85d698_flink-test-ingest-0.1.8-SNAPSHOT.jar

JAR to GUI. Also on the side note, maybe another question - is it possible that once you upload JAR to Flink, that it does not create random prefix for that JAR file locally? E.g. in my case, uploaded JAR local file name is actually

84c65347-e334-4811-869f-8f4f0b85d698_flink-test-ingest-0.1.8-SNAPSHOT.jar

And once I create a job from that JAR, it create another local file named

<random_hash>_84c65347-e334-4811-869f-8f4f0b85d698_flink-test-ingest-0.1.8-SNAPSHOT.jar

. It is something I can live with, but it would be way cleaner if I can get rid of duplication there.

Hey! The session job logic runs with the basic assumption that the jar in the

jarURI

field is not accessible on the session cluster so it has to upload it. Then it uploads it with a random prefix to avoid collision in case you are tring to submit 2 jobs with 2 different jars with the same name

Thanks for fast response 🙂 Yeah, it would be - I am mounting same NFS to both operator (so that it can create

FlinkSessionJob

as well as into Flink cluster itself.

The Flink session cluster mechanism itself doesn’t support submitting a job from a locally available jar I believe. So this is not really an operator limitation, currently in Flink itself you always have to “upload” the jar to the session before running

Thing is if I create new job via GUI, there is no duplication there, only in case I create job using CRD

The best we could get from the operator side is to upload “once” but we would need a new mechanism to allow you to define some id, instead of the name or config to not upload the same name twice

Hm, since once you upload JAR via Flink (using /jars/upload API endpoint), it generates random uuid as a file prefix (e.g. in my case

84c65347-e334-4811-869f-8f4f0b85d698_flink-test-ingest-0.1.8-SNAPSHOT.jar

, where

84c65347-e334-4811-869f-8f4f0b85d698

is uuid auto-generated by Flink during JAR upload. Would not that be sufficient? Just for my understanding (since I am pretty new to this) - does operator actually upload JAR to session cluster once you create

FlinkSessionJob

? If that is the case, I understand what you are explaining to me yes. And I guess there is no way to say to operator - just use JAR that is “local” to session cluster right?

yes the operator uploads the jar

Ah yes, it makes sense - I always thought that job is done from cluster point-of-view, but of course, it is from operator’s PoV, so it makes sense. I will give it a try to add jobs to cluster classpath, that will make things cleaner 🙂 And I guess possibility to let session decide “which local JAR to use” does not make really sense, since then operator cannot reconcile that object properly (since that part is no longer under it’s control)

It would make sense to support jars locally from the session cluster but that would have to be implemented on the Flink side first

But is not that already the case? E.g. before I deployed Flink cluster using flink binary. Then I uploaded JAR file to /jars/upload endpoint and used those JARs when creating flink jobs (during upload of course there was uuid that was appended to file name). Maybe I am just misunderstanding what you are trying to explain to me, sorry for that 🙂

Then I uploaded JAR file to /jars/upload endpoint

you were not using local files, you uploaded them

Yeah, that is the thing - I have GitHub actions pipeline that uploads new version of JAR to Flink cluster and then I plan to use that one in

FlinkSessionJob

specification 🙂

your best bet is to open a JIRA and maybe even work on it yourself. Someone might pick it up but if this feature is not a priority for some contributor/user it may not be done for a while

Perfect, thank you for your support 🙂

Perfect 🙂 I am used to work on operators using Go, I will try to get my head around this one (first one for me written in Java) 🙂