Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
Message-ID: <55A77307.5020105@tu-ilmenau.de>
Date: Thu, 16 Jul 2015 11:01:59 +0200
From: Philipp Goetze <philipp.goetze@tu-ilmenau.de>
User-Agent: Mozilla/5.0 (X11; Linux x86_64;
 rv:31.0) Gecko/20100101 Thunderbird/31.7.0
MIME-Version: 1.0
To: user@flink.apache.org
Subject: Submitting jobs from within Scala code
Content-Type: multipart/alternative;
 boundary="------------020603050005040905030004"

This is a multi-part message in MIME format.
--------------020603050005040905030004
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 8bit

Hi community,

in our project we try to submit built Flink programs to the jobmanager 
from within Scala code. The test program is executed correctly when 
submitted via the wrapper script "bin/flink run ..." and also with the 
webclient. But when executed from within the Scala code nothing seems to 
happen, but the following warning is found in the log:

10:47:18,153 WARN  akka.remote.ReliableDeliverySupervisor                        - Association with remote system [akka.tcp://flink@127.0.0.1:34074] has failed, address is now gated for [5000] ms. Reason is: [org.apache.flink.runtime.jobgraph.AbstractJobVertex]

Our submit method looks like that:

  def submitJar(master: String, path: String, className: String, args: String*) = {

     val file = new File(path)

     val parallelism = 1

     val wait = true

     try {

       val program = new PackagedProgram(file, className, args:_*)

       val jobManagerAddress = getInetFromHostport(master)

       val client = new Client(jobManagerAddress, new Configuration(), program.getUserCodeClassLoader(), 1)

       println("Executing " + path);

       client.run(program, parallelism, wait);

     } catch {

       case e: ProgramInvocationException => e.printStackTrace()

     }

   }


I took this as a reference: 
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/How-to-submit-flink-jars-from-plain-Java-programs-td656.html

I hope you can help.

Best Regards,
Philipp Götze

--------------020603050005040905030004
Content-Type: text/html; charset=utf-8
Content-Transfer-Encoding: 8bit

<html>
  <head>

    <meta http-equiv="content-type" content="text/html; charset=utf-8">
  </head>
  <body bgcolor="#FFFFFF" text="#000000">
    Hi community,<br>
    <br>
    in our project we try to submit built Flink programs to the
    jobmanager from within Scala code. The test program is executed
    correctly when submitted via the wrapper script "bin/flink run ..."
    and also with the webclient. But when executed from within the Scala
    code nothing seems to happen, but the following warning is found in
    the log:<br>
    <meta http-equiv="content-type" content="text/html; charset=utf-8">
    <pre style="color: rgb(0, 0, 0); font-style: normal; font-variant: normal; font-weight: normal; letter-spacing: normal; line-height: normal; orphans: auto; text-align: start; text-indent: 0px; text-transform: none; widows: 1; word-spacing: 0px; -webkit-text-stroke-width: 0px; word-wrap: break-word; white-space: pre-wrap;">10:47:18,153 WARN  akka.remote.ReliableDeliverySupervisor                        - Association with remote system [<a class="moz-txt-link-abbreviated" href="mailto:akka.tcp://flink@127.0.0.1:34074">akka.tcp://flink@127.0.0.1:34074</a>] has failed, address is now gated for [5000] ms. Reason is: [org.apache.flink.runtime.jobgraph.AbstractJobVertex]

</pre>
    Our submit method looks like that:<br>
    <pre> def submitJar(master: String, path: String, className: String, args: String*) = { </pre>
    <pre>    val file = new File(path)</pre>
    <pre>    val parallelism = 1 </pre>
    <pre>    val wait = true</pre>
    <pre>    try { </pre>
    <pre>      val program = new PackagedProgram(file, className, args:_*)</pre>
    <pre>      val jobManagerAddress = getInetFromHostport(master)</pre>
    <pre>      val client = new Client(jobManagerAddress, new Configuration(), program.getUserCodeClassLoader(), 1)  </pre>
    <pre>      println("Executing " + path); </pre>
    <pre>      client.run(program, parallelism, wait); </pre>
    <pre>    } catch {</pre>
    <pre>      case e: ProgramInvocationException =&gt; e.printStackTrace()</pre>
    <pre>    }   </pre>
    <pre>  }   </pre>
    <br>
    I took this as a reference: <a
href="http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/How-to-submit-flink-jars-from-plain-Java-programs-td656.html">http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/How-to-submit-flink-jars-from-plain-Java-programs-td656.html</a><br>
    <br>
    I hope you can help.<br>
    <br>
    Best Regards,<br>
    Philipp Götze<br>
  </body>
</html>

--------------020603050005040905030004--