title: No translator error while running Apache Beam job on Flink cluster
body: <p>I created a very simple apache beam job for test, it is written in scala and looks like this:</p>
  
  <pre><code>object Test {
    def main(args: Array[String]): Unit = {
      val options = PipelineOptionsFactory.fromArgs(args: _*).create()
      val p = Pipeline.create(options)
  
      println(s"--------&gt; $options")
  
      val printDoFn = new DoFn[String, Void] {
        @ProcessElement
        def processElement(c: ProcessContext): Unit = {
          val e = c.element()
          logger.info(e)
          println(s"===&gt; $e")
        }
      }
  
      p.apply(Create.of[String]("A", "B", "CCC"))
        .apply(ParDo.of(printDoFn))
  
      p.run()
    }
  }
  </code></pre>
  
  <p>Now I deployed a flink cluster with the official flink docker image.</p>
  
  <p>I created a uber-jar of my test program using maven shaded plugin.</p>
  
  <p>I uploaded this uber-jar with the web UI interface of Job Manager.</p>
  
  <p>I login into the JobManager machine, and find the uploaded uber-jar, and I run the job with:</p>
  
  <pre><code>flink run -c myapps.Test \
   ./52649b36-aa57-4f2b-95c7-2552fd737ea6_pipeline_beam-1.0.0-SNAPSHOT.jar \
   --runner=FlinkRunner
  </code></pre>
  
  <p>But I got this error:</p>
  
  <pre><code>org.apache.flink.client.program.ProgramInvocationException: The main method caused an error.
      at org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:545)
      at org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:420)
      at org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:404)
      at org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:798)
      at org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:289)
      at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:215)
      at org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1035)
      at org.apache.flink.client.cli.CliFrontend.lambda$main$9(CliFrontend.java:1111)
      at java.security.AccessController.doPrivileged(Native Method)
      at javax.security.auth.Subject.doAs(Subject.java:422)
      at org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1836)
      at org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41)
      at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1111)
  Caused by: java.lang.IllegalStateException: No translator known for org.apache.beam.sdk.io.Read$Bounded
      at org.apache.beam.runners.core.construction.PTransformTranslation.urnForTransform(PTransformTranslation.java:164)
      at org.apache.beam.runners.flink.FlinkBatchPipelineTranslator.visitPrimitiveTransform(FlinkBatchPipelineTranslator.java:93)
      at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:657)
      at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
      at org.apache.beam.sdk.runners.TransformHierarchy$Node.visit(TransformHierarchy.java:649)
      at org.apache.beam.sdk.runners.TransformHierarchy$Node.access$600(TransformHierarchy.java:311)
      at org.apache.beam.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:245)
      at org.apache.beam.sdk.Pipeline.traverseTopologically(Pipeline.java:458)
      at org.apache.beam.runners.flink.FlinkPipelineTranslator.translate(FlinkPipelineTranslator.java:38)
      at org.apache.beam.runners.flink.FlinkBatchPipelineTranslator.translate(FlinkBatchPipelineTranslator.java:49)
      at org.apache.beam.runners.flink.FlinkPipelineExecutionEnvironment.translate(FlinkPipelineExecutionEnvironment.java:119)
      at org.apache.beam.runners.flink.FlinkRunner.run(FlinkRunner.java:110)
      at org.apache.beam.sdk.Pipeline.run(Pipeline.java:313)
      at org.apache.beam.sdk.Pipeline.run(Pipeline.java:299)
      ...
  </code></pre>
  
  <p>I think the key error is: <code>No translator known for org.apache.beam.sdk.io.Read$Bounded</code></p>
  
  <p>I compiled my program with apache beam 2.7.0, and from the flink runner page: <a href="https://beam.apache.org/documentation/runners/flink/" rel="nofollow noreferrer">https://beam.apache.org/documentation/runners/flink/</a> , I deployed flink 1.5.5 version, with the flink official image: <code>flink:1.5.5-hadoop28-scala_2.11-alpine</code></p>
  
  <p>I couldn't find any useful information on Google. </p>