[GitHub] incubator-flink pull request: Do not limit the client's heap space...

[zentol]

GitHub user rmetzger opened a pull request:

    https://github.com/apache/incubator-flink/pull/18

    Do not limit the client's heap space (See FLINK-934)

    Similar to  FLINK-934 and https://github.com/apache/incubator-flink/pull/14 we should not limit the heap space of the job submission client to 512MB.
   
    A user actually had the following exception while using Flink:
    ```
    java.lang.OutOfMemoryError: Java heap space
    at eu.stratosphere.runtime.io.serialization.DataOutputSerializer.resize(DataOutputSerializer.java:243)
    at eu.stratosphere.runtime.io.serialization.DataOutputSerializer.write(DataOutputSerializer.java:87)
    at eu.stratosphere.nephele.jobgraph.JobGraph.writeRequiredJarFiles(JobGraph.java:706)
    at eu.stratosphere.nephele.jobgraph.JobGraph.write(JobGraph.java:641)
    at eu.stratosphere.nephele.ipc.RPC$Invocation.write(RPC.java:153)
    at eu.stratosphere.nephele.ipc.Client$Connection.sendParam(Client.java:469)
    at eu.stratosphere.nephele.ipc.Client.call(Client.java:689)
    at eu.stratosphere.nephele.ipc.RPC$Invoker.invoke(RPC.java:250)
    at com.sun.proxy.$Proxy1.submitJob(Unknown Source)
    at eu.stratosphere.nephele.client.JobClient.submitJobAndWait(JobClient.java:258)
    at eu.stratosphere.client.program.Client.run(Client.java:284)
    at eu.stratosphere.client.program.Client.run(Client.java:268)
    at eu.stratosphere.client.program.Client.run(Client.java:262)
    at eu.stratosphere.client.program.ContextEnvironment.execute(ContextEnvironment.java:50)
    at eu.stratosphere.example.java.wordcount.WordCount.main(WordCount.java:79)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:483)
    at eu.stratosphere.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:384)
    at eu.stratosphere.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:302)
    at eu.stratosphere.client.program.Client.run(Client.java:220)
    at eu.stratosphere.client.CliFrontend.executeProgram(CliFrontend.java:327)
    at eu.stratosphere.client.CliFrontend.run(CliFrontend.java:314)
    at eu.stratosphere.client.CliFrontend.parseParameters(CliFrontend.java:927)
    at eu.stratosphere.client.CliFrontend.main(CliFrontend.java:951)
    ```
   
    I verified that the change resolves the problem.

You can merge this pull request into a Git repository by running:

    $ git pull https://github.com/rmetzger/incubator-flink stratosphere_client_fix

Alternatively you can review and apply these changes as the patch at:

    https://github.com/apache/incubator-flink/pull/18.patch

To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:

    This closes #18
   
----
commit 8e1bff41a21d75f5e7dc78fbc5c30835a3802781
Author: Robert Metzger <[hidden email]>
Date:   2014-06-13T19:40:34Z

    Do not limit the client's heap space (See FLINK-934) [ci skip]

----


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[Ufuk Celebi]

+1

But imo this will soon be subsumed by the upcoming RPC rework.

Sent from my iPhone

> On 13 Jun 2014, at 21:44, rmetzger <[hidden email]> wrote:
>
> GitHub user rmetzger opened a pull request:
>
>    https://github.com/apache/incubator-flink/pull/18
>
>    Do not limit the client's heap space (See FLINK-934)
>
>    Similar to  FLINK-934 and https://github.com/apache/incubator-flink/pull/14 we should not limit the heap space of the job submission client to 512MB.
>
>    A user actually had the following exception while using Flink:
>    ```
>    java.lang.OutOfMemoryError: Java heap space
>        at eu.stratosphere.runtime.io.serialization.DataOutputSerializer.resize(DataOutputSerializer.java:243)
>        at eu.stratosphere.runtime.io.serialization.DataOutputSerializer.write(DataOutputSerializer.java:87)
>        at eu.stratosphere.nephele.jobgraph.JobGraph.writeRequiredJarFiles(JobGraph.java:706)
>        at eu.stratosphere.nephele.jobgraph.JobGraph.write(JobGraph.java:641)
>        at eu.stratosphere.nephele.ipc.RPC$Invocation.write(RPC.java:153)
>        at eu.stratosphere.nephele.ipc.Client$Connection.sendParam(Client.java:469)
>        at eu.stratosphere.nephele.ipc.Client.call(Client.java:689)
>        at eu.stratosphere.nephele.ipc.RPC$Invoker.invoke(RPC.java:250)
>        at com.sun.proxy.$Proxy1.submitJob(Unknown Source)
>        at eu.stratosphere.nephele.client.JobClient.submitJobAndWait(JobClient.java:258)
>        at eu.stratosphere.client.program.Client.run(Client.java:284)
>        at eu.stratosphere.client.program.Client.run(Client.java:268)
>        at eu.stratosphere.client.program.Client.run(Client.java:262)
>        at eu.stratosphere.client.program.ContextEnvironment.execute(ContextEnvironment.java:50)
>        at eu.stratosphere.example.java.wordcount.WordCount.main(WordCount.java:79)
>        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
>        at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
>        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
>        at java.lang.reflect.Method.invoke(Method.java:483)
>        at eu.stratosphere.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:384)
>        at eu.stratosphere.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:302)
>        at eu.stratosphere.client.program.Client.run(Client.java:220)
>        at eu.stratosphere.client.CliFrontend.executeProgram(CliFrontend.java:327)
>        at eu.stratosphere.client.CliFrontend.run(CliFrontend.java:314)
>        at eu.stratosphere.client.CliFrontend.parseParameters(CliFrontend.java:927)
>        at eu.stratosphere.client.CliFrontend.main(CliFrontend.java:951)
>    ```
>
>    I verified that the change resolves the problem.
>
> You can merge this pull request into a Git repository by running:
>
>    $ git pull https://github.com/rmetzger/incubator-flink stratosphere_client_fix
>
> Alternatively you can review and apply these changes as the patch at:
>
>    https://github.com/apache/incubator-flink/pull/18.patch
>
> To close this pull request, make a commit to your master/trunk branch
> with (at least) the following in the commit message:
>
>    This closes #18
>
> ----
> commit 8e1bff41a21d75f5e7dc78fbc5c30835a3802781
> Author: Robert Metzger <[hidden email]>
> Date:   2014-06-13T19:40:34Z
>
>    Do not limit the client's heap space (See FLINK-934) [ci skip]
>
> ----
>
>
> ---
> If your project is set up for it, you can reply to this email and have your
> reply appear on GitHub as well. If your project does not have this feature
> enabled and wishes so, or if the feature is enabled but not working, please
> contact infrastructure at [hidden email] or file a JIRA ticket
> with INFRA.
> ---

[Robert Metzger]

Thats good. We have the same problem on the receiver side. I assume the new
RPC service is not transferring the user-jar by first allocating it as a
whole in memory.


On Fri, Jun 13, 2014 at 9:48 PM, Ufuk Celebi <[hidden email]> wrote:

> +1
>
> But imo this will soon be subsumed by the upcoming RPC rework.
>
> Sent from my iPhone
>
> > On 13 Jun 2014, at 21:44, rmetzger <[hidden email]> wrote:
> >
> > GitHub user rmetzger opened a pull request:
> >
> >    https://github.com/apache/incubator-flink/pull/18
> >
> >    Do not limit the client's heap space (See FLINK-934)
> >
> >    Similar to  FLINK-934 and
> https://github.com/apache/incubator-flink/pull/14 we should not limit the
> heap space of the job submission client to 512MB.
> >
> >    A user actually had the following exception while using Flink:
> >    ```
> >    java.lang.OutOfMemoryError: Java heap space
> >        at
> eu.stratosphere.runtime.io.serialization.DataOutputSerializer.resize(DataOutputSerializer.java:243)
> >        at
> eu.stratosphere.runtime.io.serialization.DataOutputSerializer.write(DataOutputSerializer.java:87)
> >        at
> eu.stratosphere.nephele.jobgraph.JobGraph.writeRequiredJarFiles(JobGraph.java:706)
> >        at
> eu.stratosphere.nephele.jobgraph.JobGraph.write(JobGraph.java:641)
> >        at eu.stratosphere.nephele.ipc.RPC$Invocation.write(RPC.java:153)
> >        at
> eu.stratosphere.nephele.ipc.Client$Connection.sendParam(Client.java:469)
> >        at eu.stratosphere.nephele.ipc.Client.call(Client.java:689)
> >        at eu.stratosphere.nephele.ipc.RPC$Invoker.invoke(RPC.java:250)
> >        at com.sun.proxy.$Proxy1.submitJob(Unknown Source)
> >        at
> eu.stratosphere.nephele.client.JobClient.submitJobAndWait(JobClient.java:258)
> >        at eu.stratosphere.client.program.Client.run(Client.java:284)
> >        at eu.stratosphere.client.program.Client.run(Client.java:268)
> >        at eu.stratosphere.client.program.Client.run(Client.java:262)
> >        at
> eu.stratosphere.client.program.ContextEnvironment.execute(ContextEnvironment.java:50)
> >        at
> eu.stratosphere.example.java.wordcount.WordCount.main(WordCount.java:79)
> >        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> >        at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> >        at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> >        at java.lang.reflect.Method.invoke(Method.java:483)
> >        at
> eu.stratosphere.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:384)
> >        at
> eu.stratosphere.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:302)
> >        at eu.stratosphere.client.program.Client.run(Client.java:220)
> >        at
> eu.stratosphere.client.CliFrontend.executeProgram(CliFrontend.java:327)
> >        at eu.stratosphere.client.CliFrontend.run(CliFrontend.java:314)
> >        at
> eu.stratosphere.client.CliFrontend.parseParameters(CliFrontend.java:927)
> >        at eu.stratosphere.client.CliFrontend.main(CliFrontend.java:951)
> >    ```
> >
> >    I verified that the change resolves the problem.
> >
> > You can merge this pull request into a Git repository by running:
> >
> >    $ git pull https://github.com/rmetzger/incubator-flink
> stratosphere_client_fix
> >
> > Alternatively you can review and apply these changes as the patch at:
> >
> >    https://github.com/apache/incubator-flink/pull/18.patch
> >
> > To close this pull request, make a commit to your master/trunk branch
> > with (at least) the following in the commit message:
> >
> >    This closes #18
> >
> > ----
> > commit 8e1bff41a21d75f5e7dc78fbc5c30835a3802781
> > Author: Robert Metzger <[hidden email]>
> > Date:   2014-06-13T19:40:34Z
> >
> >    Do not limit the client's heap space (See FLINK-934) [ci skip]
> >
> > ----
> >
> >
> > ---
> > If your project is set up for it, you can reply to this email and have
> your
> > reply appear on GitHub as well. If your project does not have this
> feature
> > enabled and wishes so, or if the feature is enabled but not working,
> please
> > contact infrastructure at [hidden email] or file a JIRA
> ticket
> > with INFRA.
> > ---
>

[zentol]

Github user StephanEwen commented on the pull request:

    https://github.com/apache/incubator-flink/pull/18#issuecomment-46120860
 
    The akka-baed RPC we are trying out right now cannot transfer large data frames out of the box. We are looking into solutions for that.
   
    Anyways, it is a flaw to allocate the entire buffer in memory before sending the on-disk jars. That needs to be solved independently of the RPC.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[Ufuk Celebi]

Could you please open a separate issue for that?

On 15 Jun 2014, at 18:49, StephanEwen <[hidden email]> wrote:

> memory

[zentol]

Github user warneke commented on the pull request:

    https://github.com/apache/incubator-flink/pull/18#issuecomment-46122924
 
    Using RPCs to distribute jar files is still the quick hack from the very early days of the project. I've already fixed this issue on several branches but I guess the change did not make it to the upstream code. I could offer to take care of this problem, either through some sort of HTTP transfer or by leveraging the distributed file system.


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[Ufuk Celebi]

On 15 Jun 2014, at 20:08, warneke <[hidden email]> wrote:

> Github user warneke commented on the pull request:
>
>    https://github.com/apache/incubator-flink/pull/18#issuecomment-46122924
>
>    Using RPCs to distribute jar files is still the quick hack from the very early days of the project. I've already fixed this issue on several branches but I guess the change did not make it to the upstream code. I could offer to take care of this problem, either through some sort of HTTP transfer or by leveraging the distributed file system.
>

Hey Daniel! Too bad that the change did not make it upstream :( What did you use in the other branches? HTTP? DFS? I would prefer HTTP over DFS. But we could also easily implement the service on top of Netty (should basically be just an outbound handler which reads the file and directly sends it out and vice versa for the inbound handler).

[Ufuk Celebi]

On 15 Jun 2014, at 20:08, warneke <[hidden email]> wrote:

> Github user warneke commented on the pull request:
>
>    https://github.com/apache/incubator-flink/pull/18#issuecomment-46122924
>
>    Using RPCs to distribute jar files is still the quick hack from the very early days of the project. I've already fixed this issue on several branches but I guess the change did not make it to the upstream code. I could offer to take care of this problem, either through some sort of HTTP transfer or by leveraging the distributed file system.

Filed in https://issues.apache.org/jira/browse/FLINK-939 and assigned to Daniel.

[zentol]

Github user StephanEwen commented on the pull request:

    https://github.com/apache/incubator-flink/pull/18#issuecomment-46301981
 
    Looks good, Will merge...


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---

[zentol]

Github user asfgit closed the pull request at:

    https://github.com/apache/incubator-flink/pull/18


---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [hidden email] or file a JIRA ticket
with INFRA.
---