<p>I'm experimenting with the new Flink Kubernetes operator and I've been able to do pretty much everything that I need besides one thing: getting a JAR file from the S3 file system.</p>
<h2>Context</h2>
<p>I have a Flink application running in a EKS cluster in AWS and have all the information saved in a S3 buckets. Things like savepoints, checkpoints, high availability and JARs files are all stored there.</p>
<p>I've been able to save the savepoints, checkpoints and high availability information in the bucket, but when trying to get the JAR file from the same bucket I get the error:
<code>Could not find a file system implementation for scheme 's3'. The scheme is directly supported by Flink through the following plugins: flink-s3-fs-hadoop, flink-s3-fs-presto.</code></p>
<p>I was able to get to <a href="https://www.mail-archive.com/user@flink.apache.org/msg48176.html" rel="nofollow noreferrer">this thread</a>, but I wasn't able to get the resource fetcher to work correctly. Also the solution is not ideal and I was searching for a more direct approach.</p>
<h2>Deployment files</h2>
<p>Here's the files that I'm deploying in the cluster:</p>
<p>deployment.yml</p>
<pre><code>apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flink-deployment
spec:
  podTemplate:
    apiVersion: v1
    kind: Pod
    metadata:
      name: pod-template
    spec:
      containers:
        - name: flink-main-container
          env:
            - name: ENABLE_BUILT_IN_PLUGINS
              value: flink-s3-fs-presto-1.15.3.jar;flink-s3-fs-hadoop-1.15.3.jar
          volumeMounts:
            - mountPath: /flink-data
              name: flink-volume
      volumes:
        - name: flink-volume
          hostPath:
            path: /tmp
            type: Directory
  image: flink:1.15
  flinkVersion: v1_15
  flinkConfiguration:
    state.checkpoints.dir: s3://kubernetes-operator/checkpoints
    state.savepoints.dir: s3://kubernetes-operator/savepoints
    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
    high-availability.storageDir: s3://kubernetes-operator/ha
  jobManager:
    resource:
      memory: &quot;2048m&quot;
      cpu: 1
  taskManager:
    resource:
      memory: &quot;2048m&quot;
      cpu: 1
  serviceAccount: flink
</code></pre>
<p>session-job.yml</p>
<pre><code>apiVersion: flink.apache.org/v1beta1
kind: FlinkSessionJob
metadata:
  name: flink-session-job
spec:
  deploymentName: flink-deployment
  job:
    jarURI: s3://kubernetes-operator/savepoints/flink.jar
    parallelism: 3
    upgradeMode: savepoint
    savepointTriggerNonce: 0
</code></pre>
<p>The Flink Kubernetes operator version that I'm using is 1.3.1</p>
<p>Is there anything that I'm missing or doing wrong?</p>

Ricardo Correia
<p>For anyone else looking this up in future, here's what I did that helped.</p>
<p>Depending on your version of the k8s operator, you may now have a <a href="https://github.com/apache/flink-kubernetes-operator/pull/609" rel="nofollow noreferrer"><code>postStart</code></a> helm chart value. As of this writing, that version is yet unreleased, but if you read the source code, all it does is inject a place in the operator deployment for you to run arbitrary code.</p>
<p>So I went into edit the k8s deployment:<code>kubectl edit deployment/flink-kubernetes-operator</code>, and added it there. Note that you might need to set the plugin folder env var. Obviously if you run <code>helm install</code> / <code>helm upgrade</code>, again, this will get erased, but hopefully they release the new helm chart with the <code>postStart</code> config soon.</p>
<pre class="lang-yaml prettyprint-override"><code># This is the values.yml file for the operator.
operatorPod:
  env:
  - name: FLINK_PLUGINS_DIR
    value: /opt/flink/plugins
</code></pre>
<pre class="lang-yaml prettyprint-override"><code># This is the deployment definition of the operator. i.e. when you do kubectl edit. 
# Look for the main container spec
# ... truncated 
    containers:
      - command:
        - /docker-entrypoint.sh
        - operator
        env:
        - name: OPERATOR_NAMESPACE
          valueFrom:
            fieldRef:
              apiVersion: v1
              fieldPath: metadata.namespace
        # there might be more things in here...
        image: ghcr.io/apache/flink-kubernetes-operator:be07be7
        imagePullPolicy: IfNotPresent
        # add this following section, making sure you choose the right version for you.
        lifecycle:
          postStart:
            exec:
              command:
              - /bin/sh
              - -c
              - 'mkdir $FLINK_PLUGINS_DIR/flink-s3-fs-hadoop &amp;&amp; curl -H &quot;Accept: application/zip&quot;
                https://repo1.maven.org/maven2/org/apache/flink/flink-s3-fs-hadoop/1.15.4/flink-s3-fs-hadoop-1.15.4.jar
                --output $FLINK_PLUGINS_DIR/flink-s3-fs-hadoop/flink-s3-fs-hadoop-1.15.4.jar'
# ... truncated
</code></pre>
<p>Then, in your FlinkDeployment yaml, all you have to do is find the main container and add these 2 env vars. Flink will automagically copy the plugin(s) (semicolon or ' ' separated) into the folder so you dont have to run anything custom.</p>
<pre class="lang-yaml prettyprint-override"><code># truncated
        - name: flink-main-container
            env:
            - name: ENABLE_BUILT_IN_PLUGINS
              value: flink-s3-fs-hadoop-1.15.4.jar
            - name: &quot;FLINK_PLUGINS_DIR&quot;
              value: &quot;/opt/flink/plugins&quot;
# truncated
</code></pre>
<p>Not entirely sure why this works with the Flink image but not on the Operator one. Also not sure why there aren't clearer docs around this.</p>
<p>Unlike the previous comments, you <em>should not</em> add <code>fs.s3a.aws.credentials.provider</code> to your config. The plugin has a default list of credentials providers that it will try based on your configuration environment. I use IAM roles, making the flink serviceaccount a trusted entity in MyCustomS3FlinkWritePolicy so I don't have to put AWS secrets in EKS.</p>
