Class not found exception in user defined 'open()' function without open() function

[Maximilian Alber]

Hi Flinksters!

I get some strange error. The compiler complains about an error inside a function, but that function is not declared by my code. So I assume it is not my bad..

The code (the alternative is just used to show that the error happens in here...):

// begin alternative /////////////////////
//index_tmp neutralize
// end alternative ///////////////////////

// begin error section ///////////////////
val index = index_tmp map {x: Vector => Tuple1(1)} sum 0

val center = X.filter(new RichFilterFunction[Vector](){
    var index: Int = 0
    //override def open(config: Configuration) = {
    //val x: Tuple1[Int] = getRuntimeContext.getBroadcastVariable("index").toList.head
    //index = 0//x._1
    //}
    def filter(x: Vector) = x.id == index
}).withBroadcastSet(index, "index")

center neutralize

The error:

 Error: The program execution failed: java.lang.Exception: The user defined 'open()' method caused an exception: Repeated instantiation of serializer failed.
at org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:488)
at org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:360)
at org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.RuntimeException: Repeated instantiation of serializer failed.
at org.apache.flink.api.java.typeutils.runtime.RuntimeStatefulSerializerFactory.getSerializer(RuntimeStatefulSerializerFactory.java:102)
at org.apache.flink.api.scala.operators.ScalaAggregateOperator$AggregatingUdf.open(ScalaAggregateOperator.java:275)
at org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:33)
at org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:484)
... 3 more
Caused by: java.lang.ClassNotFoundException: bumpboost.BumpBoost$$anon$87$$anon$88
at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
at java.security.AccessController.doPrivileged(Native Method)
at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
at java.lang.Class.forName0(Native Method)
at java.lang.Class.forName(Class.java:274)
at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
at org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:57)
at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:274)
at org.apache.flink.api.java.typeutils.runtime.RuntimeStatefulSerializerFactory.getSerializer(RuntimeStatefulSerializerFactory.java:100)
... 6 more

I append the source code, jar and the input files.

May call it with these params(exchange in_file, random_file, width_candidates_file):

 'run', '-v', 'bump_boost-0.1.jar', '-c', 'bumpboost.Job', 'in_file=/tmp/tmpKd0Pq0', 'out_file=/tmp/tmpDq6EqG', 'random_file=/tmp/tmpZWiEdt', 'dimensions=1', 'N=100', 'width_candidates_file=/tmp/tmpcLVUeQ', 'N_width_candidates=50', 'iterations=30', 'multi_bump_boost=0', 'gradient_descent_iterations=30', 'cache=False', 'start_width=1.0', 'min_width=-4', 'max_width=6', 'min_width_update=1e-08', 'max_width_update=10'

Cheers

Max

[Aljoscha Krettek]

I'm looking into it.


On Thu, Dec 11, 2014 at 10:48 AM, Maximilian Alber
<[hidden email]> wrote:

> Hi Flinksters!
>
> I get some strange error. The compiler complains about an error inside a
> function, but that function is not declared by my code. So I assume it is
> not my bad..
>
> The code (the alternative is just used to show that the error happens in
> here...):
>
> // begin alternative /////////////////////
> //index_tmp neutralize
> // end alternative ///////////////////////
>
> // begin error section ///////////////////
> val index = index_tmp map {x: Vector => Tuple1(1)} sum 0
>
> val center = X.filter(new RichFilterFunction[Vector](){
>     var index: Int = 0
>     //override def open(config: Configuration) = {
>     //val x: Tuple1[Int] =
> getRuntimeContext.getBroadcastVariable("index").toList.head
>     //index = 0//x._1
>     //}
>     def filter(x: Vector) = x.id == index
> }).withBroadcastSet(index, "index")
>
> center neutralize
>
> The error:
>
>  Error: The program execution failed: java.lang.Exception: The user defined
> 'open()' method caused an exception: Repeated instantiation of serializer
> failed.
> at
> org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:488)
> at
> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:360)
> at
> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
> at java.lang.Thread.run(Thread.java:745)
> Caused by: java.lang.RuntimeException: Repeated instantiation of serializer
> failed.
> at
> org.apache.flink.api.java.typeutils.runtime.RuntimeStatefulSerializerFactory.getSerializer(RuntimeStatefulSerializerFactory.java:102)
> at
> org.apache.flink.api.scala.operators.ScalaAggregateOperator$AggregatingUdf.open(ScalaAggregateOperator.java:275)
> at
> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:33)
> at
> org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:484)
> ... 3 more
> Caused by: java.lang.ClassNotFoundException:
> bumpboost.BumpBoost$$anon$87$$anon$88
> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
> at java.security.AccessController.doPrivileged(Native Method)
> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
> at java.lang.Class.forName0(Native Method)
> at java.lang.Class.forName(Class.java:274)
> at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
> at
> org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:57)
> at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
> at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
> at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
> at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
> at
> org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:274)
> at
> org.apache.flink.api.java.typeutils.runtime.RuntimeStatefulSerializerFactory.getSerializer(RuntimeStatefulSerializerFactory.java:100)
> ... 6 more
>
>
> I append the source code, jar and the input files.
>
> May call it with these params(exchange in_file, random_file,
> width_candidates_file):
>
>  'run', '-v', 'bump_boost-0.1.jar', '-c', 'bumpboost.Job',
> 'in_file=/tmp/tmpKd0Pq0', 'out_file=/tmp/tmpDq6EqG',
> 'random_file=/tmp/tmpZWiEdt', 'dimensions=1', 'N=100',
> 'width_candidates_file=/tmp/tmpcLVUeQ', 'N_width_candidates=50',
> 'iterations=30', 'multi_bump_boost=0', 'gradient_descent_iterations=30',
> 'cache=False', 'start_width=1.0', 'min_width=-4', 'max_width=6',
> 'min_width_update=1e-08', 'max_width_update=10'
>
> Cheers
> Max

[Aljoscha Krettek]

Hi,
I found the issue and have a pending pull request for the fix. I'll
write here again as soon as it is in master. You would then have to
switch to 0.8-incubating-SNAPSHOT for your flink dependencies in your
POM.

Cheers,
Aljoscha

On Thu, Dec 11, 2014 at 11:37 AM, Aljoscha Krettek <[hidden email]> wrote:

> I'm looking into it.
>
>
> On Thu, Dec 11, 2014 at 10:48 AM, Maximilian Alber
> <[hidden email]> wrote:
>> Hi Flinksters!
>>
>> I get some strange error. The compiler complains about an error inside a
>> function, but that function is not declared by my code. So I assume it is
>> not my bad..
>>
>> The code (the alternative is just used to show that the error happens in
>> here...):
>>
>> // begin alternative /////////////////////
>> //index_tmp neutralize
>> // end alternative ///////////////////////
>>
>> // begin error section ///////////////////
>> val index = index_tmp map {x: Vector => Tuple1(1)} sum 0
>>
>> val center = X.filter(new RichFilterFunction[Vector](){
>>     var index: Int = 0
>>     //override def open(config: Configuration) = {
>>     //val x: Tuple1[Int] =
>> getRuntimeContext.getBroadcastVariable("index").toList.head
>>     //index = 0//x._1
>>     //}
>>     def filter(x: Vector) = x.id == index
>> }).withBroadcastSet(index, "index")
>>
>> center neutralize
>>
>> The error:
>>
>>  Error: The program execution failed: java.lang.Exception: The user defined
>> 'open()' method caused an exception: Repeated instantiation of serializer
>> failed.
>> at
>> org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:488)
>> at
>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:360)
>> at
>> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
>> at java.lang.Thread.run(Thread.java:745)
>> Caused by: java.lang.RuntimeException: Repeated instantiation of serializer
>> failed.
>> at
>> org.apache.flink.api.java.typeutils.runtime.RuntimeStatefulSerializerFactory.getSerializer(RuntimeStatefulSerializerFactory.java:102)
>> at
>> org.apache.flink.api.scala.operators.ScalaAggregateOperator$AggregatingUdf.open(ScalaAggregateOperator.java:275)
>> at
>> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:33)
>> at
>> org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:484)
>> ... 3 more
>> Caused by: java.lang.ClassNotFoundException:
>> bumpboost.BumpBoost$$anon$87$$anon$88
>> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
>> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
>> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
>> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
>> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
>> at java.lang.Class.forName0(Native Method)
>> at java.lang.Class.forName(Class.java:274)
>> at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
>> at
>> org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:57)
>> at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
>> at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
>> at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
>> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>> at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
>> at
>> org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:274)
>> at
>> org.apache.flink.api.java.typeutils.runtime.RuntimeStatefulSerializerFactory.getSerializer(RuntimeStatefulSerializerFactory.java:100)
>> ... 6 more
>>
>>
>> I append the source code, jar and the input files.
>>
>> May call it with these params(exchange in_file, random_file,
>> width_candidates_file):
>>
>>  'run', '-v', 'bump_boost-0.1.jar', '-c', 'bumpboost.Job',
>> 'in_file=/tmp/tmpKd0Pq0', 'out_file=/tmp/tmpDq6EqG',
>> 'random_file=/tmp/tmpZWiEdt', 'dimensions=1', 'N=100',
>> 'width_candidates_file=/tmp/tmpcLVUeQ', 'N_width_candidates=50',
>> 'iterations=30', 'multi_bump_boost=0', 'gradient_descent_iterations=30',
>> 'cache=False', 'start_width=1.0', 'min_width=-4', 'max_width=6',
>> 'min_width_update=1e-08', 'max_width_update=10'
>>
>> Cheers
>> Max

[Maximilian Alber]

Thanks!

I already do.

Cheers,

Max

On Thu, Dec 11, 2014 at 12:21 PM, Aljoscha Krettek <[hidden email]> wrote:

Hi,
I found the issue and have a pending pull request for the fix. I'll
write here again as soon as it is in master. You would then have to
switch to 0.8-incubating-SNAPSHOT for your flink dependencies in your
POM.

Cheers,
Aljoscha

On Thu, Dec 11, 2014 at 11:37 AM, Aljoscha Krettek <[hidden email]> wrote:
> I'm looking into it.
>
>
> On Thu, Dec 11, 2014 at 10:48 AM, Maximilian Alber
> <[hidden email]> wrote:
>> Hi Flinksters!
>>
>> I get some strange error. The compiler complains about an error inside a
>> function, but that function is not declared by my code. So I assume it is
>> not my bad..
>>
>> The code (the alternative is just used to show that the error happens in
>> here...):
>>
>> // begin alternative /////////////////////
>> //index_tmp neutralize
>> // end alternative ///////////////////////
>>
>> // begin error section ///////////////////
>> val index = index_tmp map {x: Vector => Tuple1(1)} sum 0
>>
>> val center = X.filter(new RichFilterFunction[Vector](){
>>     var index: Int = 0
>>     //override def open(config: Configuration) = {
>>     //val x: Tuple1[Int] =
>> getRuntimeContext.getBroadcastVariable("index").toList.head
>>     //index = 0//x._1
>>     //}
>>     def filter(x: Vector) = x.id == index
>> }).withBroadcastSet(index, "index")
>>
>> center neutralize
>>
>> The error:
>>
>>  Error: The program execution failed: java.lang.Exception: The user defined
>> 'open()' method caused an exception: Repeated instantiation of serializer
>> failed.
>> at
>> org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:488)
>> at
>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:360)
>> at
>> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
>> at java.lang.Thread.run(Thread.java:745)
>> Caused by: java.lang.RuntimeException: Repeated instantiation of serializer
>> failed.
>> at
>> org.apache.flink.api.java.typeutils.runtime.RuntimeStatefulSerializerFactory.getSerializer(RuntimeStatefulSerializerFactory.java:102)
>> at
>> org.apache.flink.api.scala.operators.ScalaAggregateOperator$AggregatingUdf.open(ScalaAggregateOperator.java:275)
>> at
>> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:33)
>> at
>> org.apache.flink.runtime.operators.RegularPactTask.run(RegularPactTask.java:484)
>> ... 3 more
>> Caused by: java.lang.ClassNotFoundException:
>> bumpboost.BumpBoost$$anon$87$$anon$88
>> at java.net.URLClassLoader$1.run(URLClassLoader.java:366)
>> at java.net.URLClassLoader$1.run(URLClassLoader.java:355)
>> at java.security.AccessController.doPrivileged(Native Method)
>> at java.net.URLClassLoader.findClass(URLClassLoader.java:354)
>> at java.lang.ClassLoader.loadClass(ClassLoader.java:425)
>> at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:308)
>> at java.lang.ClassLoader.loadClass(ClassLoader.java:358)
>> at java.lang.Class.forName0(Native Method)
>> at java.lang.Class.forName(Class.java:274)
>> at java.io.ObjectInputStream.resolveClass(ObjectInputStream.java:625)
>> at
>> org.apache.flink.util.InstantiationUtil$ClassLoaderObjectInputStream.resolveClass(InstantiationUtil.java:57)
>> at java.io.ObjectInputStream.readNonProxyDesc(ObjectInputStream.java:1612)
>> at java.io.ObjectInputStream.readClassDesc(ObjectInputStream.java:1517)
>> at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:1771)
>> at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1350)
>> at java.io.ObjectInputStream.readObject(ObjectInputStream.java:370)
>> at
>> org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:274)
>> at
>> org.apache.flink.api.java.typeutils.runtime.RuntimeStatefulSerializerFactory.getSerializer(RuntimeStatefulSerializerFactory.java:100)
>> ... 6 more
>>
>>
>> I append the source code, jar and the input files.
>>
>> May call it with these params(exchange in_file, random_file,
>> width_candidates_file):
>>
>>  'run', '-v', 'bump_boost-0.1.jar', '-c', 'bumpboost.Job',
>> 'in_file=/tmp/tmpKd0Pq0', 'out_file=/tmp/tmpDq6EqG',
>> 'random_file=/tmp/tmpZWiEdt', 'dimensions=1', 'N=100',
>> 'width_candidates_file=/tmp/tmpcLVUeQ', 'N_width_candidates=50',
>> 'iterations=30', 'multi_bump_boost=0', 'gradient_descent_iterations=30',
>> 'cache=False', 'start_width=1.0', 'min_width=-4', 'max_width=6',
>> 'min_width_update=1e-08', 'max_width_update=10'
>>
>> Cheers
>> Max