<p>I am using the WordCountProg from the tutorial on <a href="https://www.tutorialspoint.com/apache_flink/apache_flink_creating_application.htm" rel="nofollow noreferrer">https://www.tutorialspoint.com/apache_flink/apache_flink_creating_application.htm</a> . The code is as follows:</p>
<p><strong>WordCountProg.java</strong></p>
<pre><code>package main.java.spendreport;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.util.Collector;

public class WordCountProg {

       // *************************************************************************
       // PROGRAM
       // *************************************************************************
       public static void main(String[] args) throws Exception {
          final ParameterTool params = ParameterTool.fromArgs(args);
          // set up the execution environment
          final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
          // make parameters available in the web interface
          env.getConfig().setGlobalJobParameters(params);
          // get input data
          DataSet&lt;String&gt; text = env.readTextFile(params.get(&quot;input&quot;));
          DataSet&lt;Tuple2&lt;String, Integer&gt;&gt; counts =
          // split up the lines in pairs (2-tuples) containing: (word,1)
          text.flatMap(new Tokenizer())
          // group by the tuple field &quot;0&quot; and sum up tuple field &quot;1&quot;
          .groupBy(0)
          .sum(1);
          // emit result
          if (params.has(&quot;output&quot;)) {
             counts.writeAsCsv(params.get(&quot;output&quot;), &quot;\n&quot;, &quot; &quot;);
             // execute program
             env.execute(&quot;WordCount Example&quot;);
          } else {
             System.out.println(&quot;Printing result to stdout. Use --output to specify output path.&quot;);
             counts.print();
          }
       }
       
       // *************************************************************************
       // USER FUNCTIONS
       // *************************************************************************
       public static final class Tokenizer implements FlatMapFunction&lt;String, Tuple2&lt;String, Integer&gt;&gt; {
          public void flatMap(String value, Collector&lt;Tuple2&lt;String, Integer&gt;&gt; out) {
             // normalize and split the line
             String[] tokens = value.toLowerCase().split(&quot;\\W+&quot;);
             // emit the pairs
             for (String token : tokens) {
                if (token.length() &gt; 0) {
                   out.collect(new Tuple2&lt;&gt;(token, 1));
                }
             }
          }
       }
    }
</code></pre>
<p>This example takes in a text file as input, provides a count for how many times a word appears on the document, and writes the results to an output file.</p>
<p>I am creating my Job Image using the following Dockerfile:</p>
<p><strong>Dockerfile</strong></p>
<pre><code>FROM flink:1.13.0-scala_2.11
WORKDIR /opt/flink/usrlib
# Create Directory for Input/Output
RUN mkdir /opt/flink/resources
COPY target/wordcount-0.0.1-SNAPSHOT.jar /opt/flink/usrlib/wordcount.jar
</code></pre>
<p>Then the yaml for my job looks as follows:</p>
<pre><code>apiVersion: batch/v1
kind: Job
metadata:
  name: flink-jobmanager
spec:
  template:
    metadata:
      labels:
        app: flink
        component: jobmanager
    spec:
      restartPolicy: OnFailure
      containers:
        - name: jobmanager
          image: docker/wordcount:latest
          imagePullPolicy: Never
          env:
          #command: [&quot;ls&quot;]
          args: [&quot;standalone-job&quot;, &quot;--job-classname&quot;, &quot;main.java.spendreport.WordCountProg&quot;, &quot;-input&quot;, &quot;/opt/flink/resources/READ.txt&quot;, &quot;-output&quot;, &quot;/opt/flink/resources/results.txt&quot;] #, &lt;optional arguments&gt;, &lt;job arguments&gt;] # optional arguments: [&quot;--job-id&quot;, &quot;&lt;job id&gt;&quot;, &quot;--fromSavepoint&quot;, &quot;/path/to/savepoint&quot;, &quot;--allowNonRestoredState&quot;]
          #args: [&quot;standalone-job&quot;, &quot;--job-classname&quot;, &quot;org.sense.flink.examples.stream.tpch.TPCHQuery03&quot;] #, &lt;optional arguments&gt;, &lt;job arguments&gt;] # optional arguments: [&quot;--job-id&quot;, &quot;&lt;job id&gt;&quot;, &quot;--fromSavepoint&quot;, &quot;/path/to/savepoint&quot;, &quot;--allowNonRestoredState&quot;]
          ports:
            - containerPort: 6123
              name: rpc
            - containerPort: 6124
              name: blob-server
            - containerPort: 8081
              name: webui
          livenessProbe:
            tcpSocket:
              port: 6123
            initialDelaySeconds: 30
            periodSeconds: 60
          volumeMounts:
            - name: job-artifacts-volume
              mountPath: /opt/flink/resources
            - name: flink-config-volume
              mountPath: /opt/flink/conf
          securityContext:
            runAsUser: 9999  # refers to user _flink_ from official flink image, change if necessary
      volumes:
        - name: flink-config-volume
          configMap:
            name: flink-config
            items:
              - key: flink-conf.yaml
                path: flink-conf.yaml
              - key: log4j-console.properties
                path: log4j-console.properties
        - name: job-artifacts-volume
          hostPath:
          # directory location on host
            path: /Users/my-user/Documents/semafor/apache_flink/PV
</code></pre>
<p>The goal is to mount /Users/my-user/Documents/semafor/apache_flink/PV where there is a READ.txt file into the pod that serves as input to the job. But when the job tries to execute, I get the following error:</p>
<pre><code>java.io.FileNotFoundException: File /opt/flink/resources/READ.txt does not exist or the user running Flink ('flink') has insufficient permissions to access it.
</code></pre>
<p>I have tried to run:</p>
<pre><code>sudo chown -R 9999:9999 /Users/my-user/Documents/semafor/apache_flink/PV 
</code></pre>
<p>Also ran chmod 777... but I get the same error.</p>
<p>I also tried copying the jar to where the READ.txt file is: <em>/Users/my-user/Documents/semafor/apache_flink/PV</em> on my local directory and mount that to /opt/flink/usrlib instead, but then I got:</p>
<pre><code>org.apache.flink.util.FlinkException: Could not find the provided job class (main.java.spendreport.WordCountProg) in the user lib directory (/opt/flink/usrlib).
</code></pre>
<p>I am not that experienced in Kubernetes or Flink, so I'm not sure if I am mounting incorrectly or if I'm doing something wrong. If you have any suggestions, please lmk. Thanks in advance.</p>

<p>If using minikube you need to first mount the volume using</p>
<pre><code>minikube mount /Users/my-user/Documents/semafor/apache_flink/PV:/tmp/PV 
</code></pre>
<p>Then use /tmp/PV in your hostPath configuration in the volumes section</p>
<p><strong>Refer to these threads:</strong>
<a href="https://stackoverflow.com/questions/60479594/minikube-volume-write-permissions">Minikube volume write permissions?</a></p>
<p><a href="https://stackoverflow.com/questions/38682114/hostpath-with-minikube-kubernetes">HostPath with minikube - Kubernetes</a></p>
