flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Matt <dromitl...@gmail.com>
Subject Re: [BUG?] Cannot Load User Class on Local Environment
Date Mon, 15 May 2017 20:05:18 GMT
Hi Till,

I just tried with Flink 1.4 by compiling the current master branch on
GitHub (as of this morning) and I still find the same problem as before. If
I'm not wrong your PR was merged already, so your fixes should be part of
the binary.

I hope you have time to have a look at the test case in [1].

Best,
Matt

[1] https://gist.github.com/17d82ee7dd921a0d649574a361cc017d

On Thu, Apr 27, 2017 at 10:09 AM, Matt <dromitlabs@gmail.com> wrote:

> Hi Till,
>
> Great! Do you know if it's planned to be included in v1.2.x or should we
> wait for v1.3? I'll give it a try as soon as it's merged.
>
> You're right about this approach launching a mini cluster on each Ignite
> node. That is intentional, as described in my previous message on the list
> [1].
>
> The idea is to collocate Flink jobs on Ignite nodes, so each dataflow only
> processes the elements stored on the local in-memory database. I get the
> impression this should be much faster than randomly picking a Flink node
> and sending all the data over the network.
>
> Any insight on this?
>
> Cheers,
> Matt
>
> [1] http://apache-flink-user-mailing-list-archive.2336050.
> n4.nabble.com/Flink-on-Ignite-Collocation-td12780.html
>
> On Thu, Apr 27, 2017 at 5:33 AM, Till Rohrmann <trohrmann@apache.org>
> wrote:
>
>> I just copied my response because my other email address is not accepted
>> on the user mailing list.
>>
>> Hi Matt,
>>
>> I think Stefan's analysis is correct. I have a PR open [1], where I fix
>> the issue with the class loader.
>>
>> As a side note, by doing what you're doing, you will spawn on each Ignite
>> node a new Flink mini cluster. These mini cluster won't communicate with
>> each other and run independently. Is this what you intend to do?
>>
>> [1] https://github.com/apache/flink/pull/3781
>>
>> Cheers,
>> Till
>>
>> On Wed, Apr 26, 2017 at 11:12 PM, Matt <dromitlabs@gmail.com> wrote:
>>
>>> Let's wait for Till then, I hope he can figure this out.
>>>
>>> On Wed, Apr 26, 2017 at 11:03 AM, Stefan Richter <
>>> s.richter@data-artisans.com> wrote:
>>>
>>>> Ok, now the question is also about what classloaders Ignite is creating
>>>> and how they are used, but the relevant code line in Flink is probably in
>>>> FlinkMiniCluster.scala, line 538 (current master):
>>>>
>>>>  try {
>>>>  JobClient.submitJobAndWait(
>>>>    clientActorSystem,
>>>>    configuration,
>>>>    leaderRetrievalService,
>>>>    jobGraph,
>>>>    timeout,
>>>>    printUpdates,
>>>>    this.getClass.getClassLoader())
>>>> } finally {
>>>>    if(!useSingleActorSystem) {
>>>>      // we have to shutdown the just created actor system
>>>>      shutdownJobClientActorSystem(clientActorSystem)
>>>>    }
>>>>  }
>>>>
>>>>
>>>> This is what is executed as part of executing a job through
>>>> LocalEnvironment. As we can see, the classloader is set to the classloader
>>>> of FlinkMiniCluster. Depending on the classloader structure inside Ignite,
>>>> this classloader might not know your user code. What you could do is
>>>> changing this line in a custom Flink build, changing line 538 for example
>>>> to Thread.currentThread().getContextClassloader() and ensuring that
>>>> the context classloader ins the runnable is a classloader that a) knows the
>>>> user code and b) is a child of the classloader that knows the Ignite and
>>>> Flink classes. Notice that this is not a general solution and should not
>>>> become a general fix.
>>>>
>>>> I have heard that Till is about to change some things about local
>>>> execution, so I included him in CC. Maybe he can provide additional hints
>>>> how your use case might be better supported in the upcoming Flink 1.3.
>>>>
>>>> Best,
>>>> Stefan
>>>>
>>>> Am 25.04.2017 um 22:50 schrieb Matt <dromitlabs@gmail.com>:
>>>>
>>>> I updated the code a little bit for clarity, now the line #56 mentioned
>>>> in my previous message is line #25.
>>>>
>>>> In summary the error I'm getting is this:
>>>>
>>>> ---
>>>> Caused by: org.apache.flink.streaming.runtime.tasks.StreamTaskException:
>>>> Cannot load user class: com.test.Test
>>>> ClassLoader info: URL ClassLoader:
>>>> Class not resolvable through given classloader.
>>>> ---
>>>>
>>>> But if I'm not wrong, after trying to load the class through
>>>> URLClassLoader, Flink should try loading it with its parent ClassLoader,
>>>> which should be the same ClassLoader that executed the environment, and it
>>>> does have access to the class.
>>>>
>>>> Not sure what is wrong.
>>>>
>>>> On Tue, Apr 25, 2017 at 5:38 PM, Matt <dromitlabs@gmail.com> wrote:
>>>>
>>>>> Hi Stefan,
>>>>>
>>>>> Check the code here: https://gist.github.com/
>>>>> 17d82ee7dd921a0d649574a361cc017d , the output is at the bottom of the
>>>>> page.
>>>>>
>>>>> Here are the results of the additional tests you mentioned:
>>>>>
>>>>> 1. I was able to instantiate an inner class (Test$Foo) inside the
>>>>> Ignite closure, no problem with that
>>>>> 2. I tried implementing SourceFunction and SinkFunction in Test
>>>>> itself, I was able to instantiate the class inside the Ignite closure
>>>>> 3. I'm not sure what you meant in this point, is it something like
>>>>> what I tried in line #56?
>>>>>
>>>>> Additionally, I tried implementing the SourceFunction and SinkFunction
>>>>> in Test$Foo with the same result: it says "Cannot load user class:
>>>>> com.test.Test$Foo"
>>>>>
>>>>> Looks like Flink is not using the correct ClassLoader. Any idea?
>>>>>
>>>>> Regards,
>>>>> Matt
>>>>>
>>>>> On Tue, Apr 25, 2017 at 7:00 AM, Stefan Richter <
>>>>> s.richter@data-artisans.com> wrote:
>>>>>
>>>>>> Hi,
>>>>>>
>>>>>> I would expect that the local environment picks up the class path
>>>>>> from the code that launched it. So I think the question is what happens
>>>>>> behind the scenes when you call ignite.compute().broadcast(runnable);
.
>>>>>> Which classes are shipped and how is the classpath build in the environment
>>>>>> that runs the code. Your example is also not fully conclusive, because
>>>>>> com.myproj.Test (which you can successfully instantiate) and
>>>>>> com.myproj.Test$1$2 (which fails) are different classes, so maybe
only the
>>>>>> outer class is shipped with the broadcast call. My theory is that
not all
>>>>>> classes are shipped (e.g. inner classes), but only Test . You could
try
>>>>>> three things to analyze to problem a little more:
>>>>>>
>>>>>> 1) Create another inner class inside Test and try if you are still
>>>>>> able to instantiate also this class via reflection.
>>>>>> 2) Let Test class itself implement the map function (avoiding the
>>>>>> usage of other/inner classes) and see if this works.
>>>>>> 3) Check and set the thread’s context classloader inside the runnable
>>>>>> to something that contains all required classes and see if this gets
picked
>>>>>> up by Flink.
>>>>>>
>>>>>> Best,
>>>>>> Stefan
>>>>>>
>>>>>> Am 25.04.2017 um 07:27 schrieb Matt <dromitlabs@gmail.com>:
>>>>>>
>>>>>> Hi all,
>>>>>>
>>>>>> I'm trying to run Flink using a local environment, but on an Ignite
>>>>>> node to achieve collocation (as mentioned in my previous message
on this
>>>>>> list).
>>>>>>
>>>>>> Have a look at the code in [1]. It's pretty simple, but I'm getting
a
>>>>>> "cannot load user class" error as shown in [2].
>>>>>>
>>>>>> If you check line #29 on the code, I'm able to create an instance
of
>>>>>> class Test, and it's the same context from which I'm creating the
Flink
>>>>>> job. Shouldn't it work provided I'm using a local environment?
>>>>>>
>>>>>> It would be really nice to be able to inject a ClassLoader into the
>>>>>> chunk of code that creates the job. Is this currently possible?
>>>>>>
>>>>>> Any fix or workaround is appreciated!
>>>>>>
>>>>>> Best,
>>>>>> Matt
>>>>>>
>>>>>> [1] https://gist.github.com/f248187b9638023b95ba8bd9d7f06215
>>>>>> [2] https://gist.github.com/796ee05425535ece1736df7b1e884cce
>>>>>>
>>>>>>
>>>>>>
>>>>>
>>>>
>>>>
>>>
>>
>

Mime
View raw message