ReduceGroup fails on server

[Arvid Heise-2]

Hi Flinker,

I'm currently desparetely trying to get a workflow to run remotely on a server. The workflow works fine in the local execution environment (both with ExecutionEnvironment.getExecutionEnvironment and ExecutionEnvironment.createLocalEnvironment(2)).

On the server, I get

01/29/2015 16:05:22:    GroupReduce (GroupReduce at de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$.main(OutbreakAnalysis.scala:79)) (1/1) switched to FAILED
java.lang.RuntimeException: : Repeated instantiation of serializer failed.
        at org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:340)
        at org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
        at java.lang.Thread.run(Thread.java:722)
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.java.typeutils.runtime.TupleComparatorBase.instantiateDeserializationUtils(TupleComparatorBase.java:267)
        at org.apache.flink.api.scala.typeutils.CaseClassComparator.duplicate(CaseClassComparator.scala:42)
        at org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:360)
        at org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:227)
        at org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:195)
        at org.apache.flink.runtime.operators.RegularPactTask.initInputLocalStrategy(RegularPactTask.java:981)
        at org.apache.flink.runtime.operators.RegularPactTask.initLocalStrategies(RegularPactTask.java:842)
        at org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:336)
        ... 2 more
Caused by: java.lang.ClassNotFoundException: de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$$anon$13$$anon$14$$anon$15

The class is of course in the submitted jar and I use the same flink and java version on both machines. Could it be that the wrong class loader is used for duplicating the comparator?

The code is available here https://github.com/SiLeBAT/Other/tree/master/de.bund.bfr.flink.outbreakanalysis .

If you need help to invoke the program, please msg me privately, so that I can send you the non-public datasets.

Thank you in advance,

Arvid

[aalexandrov]

have you tried declaring your UDF classes (e.g. TotalRankDistribution) as
static?

2015-01-29 16:14 GMT+01:00 Arvid Heise <[hidden email]>:

> Hi Flinker,
>
> I'm currently desparetely trying to get a workflow to run remotely on a
> server. The workflow works fine in the local execution environment (both
> with ExecutionEnvironment.getExecutionEnvironment and
> ExecutionEnvironment.createLocalEnvironment(2)).
>
> On the server, I get
>
> 01/29/2015 16:05:22:    GroupReduce (GroupReduce at
> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$.main(OutbreakAnalysis.scala:79))
> (1/1) switched to FAILED
> java.lang.RuntimeException: : Repeated instantiation of serializer failed.
>         at
> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:340)
>         at
> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
>         at java.lang.Thread.run(Thread.java:722)
> 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.java.typeutils.runtime.TupleComparatorBase.instantiateDeserializationUtils(TupleComparatorBase.java:267)
>         at
> org.apache.flink.api.scala.typeutils.CaseClassComparator.duplicate(CaseClassComparator.scala:42)
>         at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:360)
>         at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:227)
>         at
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:195)
>         at
> org.apache.flink.runtime.operators.RegularPactTask.initInputLocalStrategy(RegularPactTask.java:981)
>         at
> org.apache.flink.runtime.operators.RegularPactTask.initLocalStrategies(RegularPactTask.java:842)
>         at
> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:336)
>         ... 2 more
> Caused by: java.lang.ClassNotFoundException:
> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$$anon$13$$anon$14$$anon$15
>
> The class is of course in the submitted jar and I use the same flink and
> java version on both machines. Could it be that the wrong class loader is
> used for duplicating the comparator?
>
> The code is available here
> https://github.com/SiLeBAT/Other/tree/master/de.bund.bfr.flink.outbreakanalysis
> .
> If you need help to invoke the program, please msg me privately, so that I
> can send you the non-public datasets.
>
> Thank you in advance,
>
> Arvid
>

[aalexandrov]

Forget what I just said, didn't realize that it's Scala :)

2015-01-29 16:24 GMT+01:00 Alexander Alexandrov <
[hidden email]>:

> have you tried declaring your UDF classes (e.g. TotalRankDistribution) as
> static?
>
> 2015-01-29 16:14 GMT+01:00 Arvid Heise <[hidden email]>:
>
>> Hi Flinker,
>>
>> I'm currently desparetely trying to get a workflow to run remotely on a
>> server. The workflow works fine in the local execution environment (both
>> with ExecutionEnvironment.getExecutionEnvironment and
>> ExecutionEnvironment.createLocalEnvironment(2)).
>>
>> On the server, I get
>>
>> 01/29/2015 16:05:22:    GroupReduce (GroupReduce at
>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$.main(OutbreakAnalysis.scala:79))
>> (1/1) switched to FAILED
>> java.lang.RuntimeException: : Repeated instantiation of serializer failed.
>>         at
>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:340)
>>         at
>> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
>>         at java.lang.Thread.run(Thread.java:722)
>> 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.java.typeutils.runtime.TupleComparatorBase.instantiateDeserializationUtils(TupleComparatorBase.java:267)
>>         at
>> org.apache.flink.api.scala.typeutils.CaseClassComparator.duplicate(CaseClassComparator.scala:42)
>>         at
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:360)
>>         at
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:227)
>>         at
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:195)
>>         at
>> org.apache.flink.runtime.operators.RegularPactTask.initInputLocalStrategy(RegularPactTask.java:981)
>>         at
>> org.apache.flink.runtime.operators.RegularPactTask.initLocalStrategies(RegularPactTask.java:842)
>>         at
>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:336)
>>         ... 2 more
>> Caused by: java.lang.ClassNotFoundException:
>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$$anon$13$$anon$14$$anon$15
>>
>> The class is of course in the submitted jar and I use the same flink and
>> java version on both machines. Could it be that the wrong class loader is
>> used for duplicating the comparator?
>>
>> The code is available here
>> https://github.com/SiLeBAT/Other/tree/master/de.bund.bfr.flink.outbreakanalysis
>> .
>> If you need help to invoke the program, please msg me privately, so that
>> I can send you the non-public datasets.
>>
>> Thank you in advance,
>>
>> Arvid
>>
>
>

[Aljoscha Krettek-2]

Hi Arvid,
this indeed looks like a ClassLoader problem. Are you running a
version that you compiled yourself? If yes, could you try replacing
line 49 in RuntimeStatefulSerializerFactory.java with this line:

this.loader = Thread.currentThread().getContextClassLoader();

Cheers,
Aljoscha

On Thu, Jan 29, 2015 at 4:25 PM, Alexander Alexandrov
<[hidden email]> wrote:

> Forget what I just said, didn't realize that it's Scala :)
>
> 2015-01-29 16:24 GMT+01:00 Alexander Alexandrov <
> [hidden email]>:
>
>> have you tried declaring your UDF classes (e.g. TotalRankDistribution) as
>> static?
>>
>> 2015-01-29 16:14 GMT+01:00 Arvid Heise <[hidden email]>:
>>
>>> Hi Flinker,
>>>
>>> I'm currently desparetely trying to get a workflow to run remotely on a
>>> server. The workflow works fine in the local execution environment (both
>>> with ExecutionEnvironment.getExecutionEnvironment and
>>> ExecutionEnvironment.createLocalEnvironment(2)).
>>>
>>> On the server, I get
>>>
>>> 01/29/2015 16:05:22:    GroupReduce (GroupReduce at
>>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$.main(OutbreakAnalysis.scala:79))
>>> (1/1) switched to FAILED
>>> java.lang.RuntimeException: : Repeated instantiation of serializer failed.
>>>         at
>>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:340)
>>>         at
>>> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
>>>         at java.lang.Thread.run(Thread.java:722)
>>> 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.java.typeutils.runtime.TupleComparatorBase.instantiateDeserializationUtils(TupleComparatorBase.java:267)
>>>         at
>>> org.apache.flink.api.scala.typeutils.CaseClassComparator.duplicate(CaseClassComparator.scala:42)
>>>         at
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:360)
>>>         at
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:227)
>>>         at
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:195)
>>>         at
>>> org.apache.flink.runtime.operators.RegularPactTask.initInputLocalStrategy(RegularPactTask.java:981)
>>>         at
>>> org.apache.flink.runtime.operators.RegularPactTask.initLocalStrategies(RegularPactTask.java:842)
>>>         at
>>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:336)
>>>         ... 2 more
>>> Caused by: java.lang.ClassNotFoundException:
>>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$$anon$13$$anon$14$$anon$15
>>>
>>> The class is of course in the submitted jar and I use the same flink and
>>> java version on both machines. Could it be that the wrong class loader is
>>> used for duplicating the comparator?
>>>
>>> The code is available here
>>> https://github.com/SiLeBAT/Other/tree/master/de.bund.bfr.flink.outbreakanalysis
>>> .
>>> If you need help to invoke the program, please msg me privately, so that
>>> I can send you the non-public datasets.
>>>
>>> Thank you in advance,
>>>
>>> Arvid
>>>
>>
>>

[Arvid Heise-2]

No I'm using the maven builds, I could try a nightly if you like.

On Thu, Jan 29, 2015 at 4:34 PM, Aljoscha Krettek <[hidden email]>
wrote:

> Hi Arvid,
> this indeed looks like a ClassLoader problem. Are you running a
> version that you compiled yourself? If yes, could you try replacing
> line 49 in RuntimeStatefulSerializerFactory.java with this line:
>
> this.loader = Thread.currentThread().getContextClassLoader();
>
> Cheers,
> Aljoscha
>
> On Thu, Jan 29, 2015 at 4:25 PM, Alexander Alexandrov
> <[hidden email]> wrote:
> > Forget what I just said, didn't realize that it's Scala :)
> >
> > 2015-01-29 16:24 GMT+01:00 Alexander Alexandrov <
> > [hidden email]>:
> >
> >> have you tried declaring your UDF classes (e.g. TotalRankDistribution)
> as
> >> static?
> >>
> >> 2015-01-29 16:14 GMT+01:00 Arvid Heise <[hidden email]>:
> >>
> >>> Hi Flinker,
> >>>
> >>> I'm currently desparetely trying to get a workflow to run remotely on a
> >>> server. The workflow works fine in the local execution environment
> (both
> >>> with ExecutionEnvironment.getExecutionEnvironment and
> >>> ExecutionEnvironment.createLocalEnvironment(2)).
> >>>
> >>> On the server, I get
> >>>
> >>> 01/29/2015 16:05:22:    GroupReduce (GroupReduce at
> >>>
> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$.main(OutbreakAnalysis.scala:79))
> >>> (1/1) switched to FAILED
> >>> java.lang.RuntimeException: : Repeated instantiation of serializer
> failed.
> >>>         at
> >>>
> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:340)
> >>>         at
> >>>
> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
> >>>         at java.lang.Thread.run(Thread.java:722)
> >>> 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.java.typeutils.runtime.TupleComparatorBase.instantiateDeserializationUtils(TupleComparatorBase.java:267)
> >>>         at
> >>>
> org.apache.flink.api.scala.typeutils.CaseClassComparator.duplicate(CaseClassComparator.scala:42)
> >>>         at
> >>>
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:360)
> >>>         at
> >>>
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:227)
> >>>         at
> >>>
> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:195)
> >>>         at
> >>>
> org.apache.flink.runtime.operators.RegularPactTask.initInputLocalStrategy(RegularPactTask.java:981)
> >>>         at
> >>>
> org.apache.flink.runtime.operators.RegularPactTask.initLocalStrategies(RegularPactTask.java:842)
> >>>         at
> >>>
> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:336)
> >>>         ... 2 more
> >>> Caused by: java.lang.ClassNotFoundException:
> >>>
> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$$anon$13$$anon$14$$anon$15
> >>>
> >>> The class is of course in the submitted jar and I use the same flink
> and
> >>> java version on both machines. Could it be that the wrong class loader
> is
> >>> used for duplicating the comparator?
> >>>
> >>> The code is available here
> >>>
> https://github.com/SiLeBAT/Other/tree/master/de.bund.bfr.flink.outbreakanalysis
> >>> .
> >>> If you need help to invoke the program, please msg me privately, so
> that
> >>> I can send you the non-public datasets.
> >>>
> >>> Thank you in advance,
> >>>
> >>> Arvid
> >>>
> >>
> >>
>

[Arvid Heise-2]

Nevermind, I'm going to build it myself and try your patch.

On Thu, Jan 29, 2015 at 4:39 PM, Arvid Heise <[hidden email]> wrote:

> No I'm using the maven builds, I could try a nightly if you like.
>
> On Thu, Jan 29, 2015 at 4:34 PM, Aljoscha Krettek <[hidden email]>
> wrote:
>
>> Hi Arvid,
>> this indeed looks like a ClassLoader problem. Are you running a
>> version that you compiled yourself? If yes, could you try replacing
>> line 49 in RuntimeStatefulSerializerFactory.java with this line:
>>
>> this.loader = Thread.currentThread().getContextClassLoader();
>>
>> Cheers,
>> Aljoscha
>>
>> On Thu, Jan 29, 2015 at 4:25 PM, Alexander Alexandrov
>> <[hidden email]> wrote:
>> > Forget what I just said, didn't realize that it's Scala :)
>> >
>> > 2015-01-29 16:24 GMT+01:00 Alexander Alexandrov <
>> > [hidden email]>:
>> >
>> >> have you tried declaring your UDF classes (e.g. TotalRankDistribution)
>> as
>> >> static?
>> >>
>> >> 2015-01-29 16:14 GMT+01:00 Arvid Heise <[hidden email]>:
>> >>
>> >>> Hi Flinker,
>> >>>
>> >>> I'm currently desparetely trying to get a workflow to run remotely on
>> a
>> >>> server. The workflow works fine in the local execution environment
>> (both
>> >>> with ExecutionEnvironment.getExecutionEnvironment and
>> >>> ExecutionEnvironment.createLocalEnvironment(2)).
>> >>>
>> >>> On the server, I get
>> >>>
>> >>> 01/29/2015 16:05:22:    GroupReduce (GroupReduce at
>> >>>
>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$.main(OutbreakAnalysis.scala:79))
>> >>> (1/1) switched to FAILED
>> >>> java.lang.RuntimeException: : Repeated instantiation of serializer
>> failed.
>> >>>         at
>> >>>
>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:340)
>> >>>         at
>> >>>
>> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
>> >>>         at java.lang.Thread.run(Thread.java:722)
>> >>> 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.java.typeutils.runtime.TupleComparatorBase.instantiateDeserializationUtils(TupleComparatorBase.java:267)
>> >>>         at
>> >>>
>> org.apache.flink.api.scala.typeutils.CaseClassComparator.duplicate(CaseClassComparator.scala:42)
>> >>>         at
>> >>>
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:360)
>> >>>         at
>> >>>
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:227)
>> >>>         at
>> >>>
>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:195)
>> >>>         at
>> >>>
>> org.apache.flink.runtime.operators.RegularPactTask.initInputLocalStrategy(RegularPactTask.java:981)
>> >>>         at
>> >>>
>> org.apache.flink.runtime.operators.RegularPactTask.initLocalStrategies(RegularPactTask.java:842)
>> >>>         at
>> >>>
>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:336)
>> >>>         ... 2 more
>> >>> Caused by: java.lang.ClassNotFoundException:
>> >>>
>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$$anon$13$$anon$14$$anon$15
>> >>>
>> >>> The class is of course in the submitted jar and I use the same flink
>> and
>> >>> java version on both machines. Could it be that the wrong class
>> loader is
>> >>> used for duplicating the comparator?
>> >>>
>> >>> The code is available here
>> >>>
>> https://github.com/SiLeBAT/Other/tree/master/de.bund.bfr.flink.outbreakanalysis
>> >>> .
>> >>> If you need help to invoke the program, please msg me privately, so
>> that
>> >>> I can send you the non-public datasets.
>> >>>
>> >>> Thank you in advance,
>> >>>
>> >>> Arvid
>> >>>
>> >>
>> >>
>>
>
>

[Arvid Heise-2]

Quickfix did not help :/
Any other idea?

On Thu, Jan 29, 2015 at 4:45 PM, Arvid Heise <[hidden email]> wrote:

> Nevermind, I'm going to build it myself and try your patch.
>
> On Thu, Jan 29, 2015 at 4:39 PM, Arvid Heise <[hidden email]>
> wrote:
>
>> No I'm using the maven builds, I could try a nightly if you like.
>>
>> On Thu, Jan 29, 2015 at 4:34 PM, Aljoscha Krettek <[hidden email]>
>> wrote:
>>
>>> Hi Arvid,
>>> this indeed looks like a ClassLoader problem. Are you running a
>>> version that you compiled yourself? If yes, could you try replacing
>>> line 49 in RuntimeStatefulSerializerFactory.java with this line:
>>>
>>> this.loader = Thread.currentThread().getContextClassLoader();
>>>
>>> Cheers,
>>> Aljoscha
>>>
>>> On Thu, Jan 29, 2015 at 4:25 PM, Alexander Alexandrov
>>> <[hidden email]> wrote:
>>> > Forget what I just said, didn't realize that it's Scala :)
>>> >
>>> > 2015-01-29 16:24 GMT+01:00 Alexander Alexandrov <
>>> > [hidden email]>:
>>> >
>>> >> have you tried declaring your UDF classes (e.g.
>>> TotalRankDistribution) as
>>> >> static?
>>> >>
>>> >> 2015-01-29 16:14 GMT+01:00 Arvid Heise <[hidden email]>:
>>> >>
>>> >>> Hi Flinker,
>>> >>>
>>> >>> I'm currently desparetely trying to get a workflow to run remotely
>>> on a
>>> >>> server. The workflow works fine in the local execution environment
>>> (both
>>> >>> with ExecutionEnvironment.getExecutionEnvironment and
>>> >>> ExecutionEnvironment.createLocalEnvironment(2)).
>>> >>>
>>> >>> On the server, I get
>>> >>>
>>> >>> 01/29/2015 16:05:22:    GroupReduce (GroupReduce at
>>> >>>
>>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$.main(OutbreakAnalysis.scala:79))
>>> >>> (1/1) switched to FAILED
>>> >>> java.lang.RuntimeException: : Repeated instantiation of serializer
>>> failed.
>>> >>>         at
>>> >>>
>>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:340)
>>> >>>         at
>>> >>>
>>> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
>>> >>>         at java.lang.Thread.run(Thread.java:722)
>>> >>> 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.java.typeutils.runtime.TupleComparatorBase.instantiateDeserializationUtils(TupleComparatorBase.java:267)
>>> >>>         at
>>> >>>
>>> org.apache.flink.api.scala.typeutils.CaseClassComparator.duplicate(CaseClassComparator.scala:42)
>>> >>>         at
>>> >>>
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:360)
>>> >>>         at
>>> >>>
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:227)
>>> >>>         at
>>> >>>
>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:195)
>>> >>>         at
>>> >>>
>>> org.apache.flink.runtime.operators.RegularPactTask.initInputLocalStrategy(RegularPactTask.java:981)
>>> >>>         at
>>> >>>
>>> org.apache.flink.runtime.operators.RegularPactTask.initLocalStrategies(RegularPactTask.java:842)
>>> >>>         at
>>> >>>
>>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:336)
>>> >>>         ... 2 more
>>> >>> Caused by: java.lang.ClassNotFoundException:
>>> >>>
>>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$$anon$13$$anon$14$$anon$15
>>> >>>
>>> >>> The class is of course in the submitted jar and I use the same flink
>>> and
>>> >>> java version on both machines. Could it be that the wrong class
>>> loader is
>>> >>> used for duplicating the comparator?
>>> >>>
>>> >>> The code is available here
>>> >>>
>>> https://github.com/SiLeBAT/Other/tree/master/de.bund.bfr.flink.outbreakanalysis
>>> >>> .
>>> >>> If you need help to invoke the program, please msg me privately, so
>>> that
>>> >>> I can send you the non-public datasets.
>>> >>>
>>> >>> Thank you in advance,
>>> >>>
>>> >>> Arvid
>>> >>>
>>> >>
>>> >>
>>>
>>
>>
>

[Arvid Heise-2]

A quick and dirty hack revealed that the loader is null at the time of
failure with and without your patch.

On Thu, Jan 29, 2015 at 5:05 PM, Arvid Heise <[hidden email]> wrote:

> Quickfix did not help :/
> Any other idea?
>
> On Thu, Jan 29, 2015 at 4:45 PM, Arvid Heise <[hidden email]>
> wrote:
>
>> Nevermind, I'm going to build it myself and try your patch.
>>
>> On Thu, Jan 29, 2015 at 4:39 PM, Arvid Heise <[hidden email]>
>> wrote:
>>
>>> No I'm using the maven builds, I could try a nightly if you like.
>>>
>>> On Thu, Jan 29, 2015 at 4:34 PM, Aljoscha Krettek <[hidden email]>
>>> wrote:
>>>
>>>> Hi Arvid,
>>>> this indeed looks like a ClassLoader problem. Are you running a
>>>> version that you compiled yourself? If yes, could you try replacing
>>>> line 49 in RuntimeStatefulSerializerFactory.java with this line:
>>>>
>>>> this.loader = Thread.currentThread().getContextClassLoader();
>>>>
>>>> Cheers,
>>>> Aljoscha
>>>>
>>>> On Thu, Jan 29, 2015 at 4:25 PM, Alexander Alexandrov
>>>> <[hidden email]> wrote:
>>>> > Forget what I just said, didn't realize that it's Scala :)
>>>> >
>>>> > 2015-01-29 16:24 GMT+01:00 Alexander Alexandrov <
>>>> > [hidden email]>:
>>>> >
>>>> >> have you tried declaring your UDF classes (e.g.
>>>> TotalRankDistribution) as
>>>> >> static?
>>>> >>
>>>> >> 2015-01-29 16:14 GMT+01:00 Arvid Heise <[hidden email]>:
>>>> >>
>>>> >>> Hi Flinker,
>>>> >>>
>>>> >>> I'm currently desparetely trying to get a workflow to run remotely
>>>> on a
>>>> >>> server. The workflow works fine in the local execution environment
>>>> (both
>>>> >>> with ExecutionEnvironment.getExecutionEnvironment and
>>>> >>> ExecutionEnvironment.createLocalEnvironment(2)).
>>>> >>>
>>>> >>> On the server, I get
>>>> >>>
>>>> >>> 01/29/2015 16:05:22:    GroupReduce (GroupReduce at
>>>> >>>
>>>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$.main(OutbreakAnalysis.scala:79))
>>>> >>> (1/1) switched to FAILED
>>>> >>> java.lang.RuntimeException: : Repeated instantiation of serializer
>>>> failed.
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:340)
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
>>>> >>>         at java.lang.Thread.run(Thread.java:722)
>>>> >>> 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.java.typeutils.runtime.TupleComparatorBase.instantiateDeserializationUtils(TupleComparatorBase.java:267)
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.api.scala.typeutils.CaseClassComparator.duplicate(CaseClassComparator.scala:42)
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:360)
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:227)
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.runtime.operators.sort.UnilateralSortMerger.<init>(UnilateralSortMerger.java:195)
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.runtime.operators.RegularPactTask.initInputLocalStrategy(RegularPactTask.java:981)
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.runtime.operators.RegularPactTask.initLocalStrategies(RegularPactTask.java:842)
>>>> >>>         at
>>>> >>>
>>>> org.apache.flink.runtime.operators.RegularPactTask.invoke(RegularPactTask.java:336)
>>>> >>>         ... 2 more
>>>> >>> Caused by: java.lang.ClassNotFoundException:
>>>> >>>
>>>> de.bund.bfr.flink.outbreakanalysis.OutbreakAnalysis$$anon$13$$anon$14$$anon$15
>>>> >>>
>>>> >>> The class is of course in the submitted jar and I use the same
>>>> flink and
>>>> >>> java version on both machines. Could it be that the wrong class
>>>> loader is
>>>> >>> used for duplicating the comparator?
>>>> >>>
>>>> >>> The code is available here
>>>> >>>
>>>> https://github.com/SiLeBAT/Other/tree/master/de.bund.bfr.flink.outbreakanalysis
>>>> >>> .
>>>> >>> If you need help to invoke the program, please msg me privately, so
>>>> that
>>>> >>> I can send you the non-public datasets.
>>>> >>>
>>>> >>> Thank you in advance,
>>>> >>>
>>>> >>> Arvid
>>>> >>>
>>>> >>
>>>> >>
>>>>
>>>
>>>
>>
>

[Stephan Ewen]

Thanks for reporting this, Arvid. I remember that we saw something similar
a while back. I will take a look at this later today (I am in PST zone
now), can hopefully fix this.

Greetings,
Stephan

[Aljoscha Krettek-2]

Of course it doesn't work. The ClassLoader is declared transient in
the serializer factory, so it is Null once the factory has been
serialized/deserialized once. I open a Jira issue:
https://issues.apache.org/jira/browse/FLINK-1463

On Thu, Jan 29, 2015 at 9:43 PM, Stephan Ewen <[hidden email]> wrote:
> Thanks for reporting this, Arvid. I remember that we saw something similar
> a while back. I will take a look at this later today (I am in PST zone
> now), can hopefully fix this.
>
> Greetings,
> Stephan

[Aljoscha Krettek-2]

Maybe we can get rid of the serializer factories altogether. We could
enhance the Serializers with a method duplicate() that does nothing
for stateless serializers and does a deep copy for stateful
serializers. This would also consolidate all the knowledge about
stateful/stateless in one place.

On Fri, Jan 30, 2015 at 11:25 AM, Aljoscha Krettek <[hidden email]> wrote:

> Of course it doesn't work. The ClassLoader is declared transient in
> the serializer factory, so it is Null once the factory has been
> serialized/deserialized once. I open a Jira issue:
> https://issues.apache.org/jira/browse/FLINK-1463
>
> On Thu, Jan 29, 2015 at 9:43 PM, Stephan Ewen <[hidden email]> wrote:
>> Thanks for reporting this, Arvid. I remember that we saw something similar
>> a while back. I will take a look at this later today (I am in PST zone
>> now), can hopefully fix this.
>>
>> Greetings,
>> Stephan

[Aljoscha Krettek-2]

We have a bit of a divide in how we handle TypeSerializer and
TypeComparator: TypeSerializer does not handle duplication but relies
on outside code (RuntimeStatefulSerializerFactory) to perform the
duplication. TypeComparator does duplication itself. There is also the
RuntimeComparatorFactory. This, however, does not perform duplication
but simply hands out the same TypeComparator instance multiple times.
With comparators, the user of the comparator is responsible for
calling duplicate() on the comparator.

I started work on a branch where I add a duplicate() method to
TypeSerializer that simply return itself for stateless serializers and
does a deep copy if it is stateful. The TypeSerializers no longer have
method isStateful() and I replaced the two serializer factories by one
factory that always returns a "duplicate" of the serializer it holds.

I think we should consolidate the two approaches. So either let the
factories handle all the duplication or let the user of the
comparator/serializer always handle duplication. I think the former is
better, since the latter makes it very easy to forget to duplicate.

On Fri, Jan 30, 2015 at 11:34 AM, Aljoscha Krettek <[hidden email]> wrote:

> Maybe we can get rid of the serializer factories altogether. We could
> enhance the Serializers with a method duplicate() that does nothing
> for stateless serializers and does a deep copy for stateful
> serializers. This would also consolidate all the knowledge about
> stateful/stateless in one place.
>
> On Fri, Jan 30, 2015 at 11:25 AM, Aljoscha Krettek <[hidden email]> wrote:
>> Of course it doesn't work. The ClassLoader is declared transient in
>> the serializer factory, so it is Null once the factory has been
>> serialized/deserialized once. I open a Jira issue:
>> https://issues.apache.org/jira/browse/FLINK-1463
>>
>> On Thu, Jan 29, 2015 at 9:43 PM, Stephan Ewen <[hidden email]> wrote:
>>> Thanks for reporting this, Arvid. I remember that we saw something similar
>>> a while back. I will take a look at this later today (I am in PST zone
>>> now), can hopefully fix this.
>>>
>>> Greetings,
>>> Stephan

[Aljoscha Krettek-2]

Hi Arvid,
I have a fix that I hope fixes your problem:
https://github.com/aljoscha/flink/tree/serializer-factories-fix

Could you try building it and running your example?

Cheers,
Aljoscha

On Fri, Jan 30, 2015 at 3:30 PM, Aljoscha Krettek <[hidden email]> wrote:

> We have a bit of a divide in how we handle TypeSerializer and
> TypeComparator: TypeSerializer does not handle duplication but relies
> on outside code (RuntimeStatefulSerializerFactory) to perform the
> duplication. TypeComparator does duplication itself. There is also the
> RuntimeComparatorFactory. This, however, does not perform duplication
> but simply hands out the same TypeComparator instance multiple times.
> With comparators, the user of the comparator is responsible for
> calling duplicate() on the comparator.
>
> I started work on a branch where I add a duplicate() method to
> TypeSerializer that simply return itself for stateless serializers and
> does a deep copy if it is stateful. The TypeSerializers no longer have
> method isStateful() and I replaced the two serializer factories by one
> factory that always returns a "duplicate" of the serializer it holds.
>
> I think we should consolidate the two approaches. So either let the
> factories handle all the duplication or let the user of the
> comparator/serializer always handle duplication. I think the former is
> better, since the latter makes it very easy to forget to duplicate.
>
> On Fri, Jan 30, 2015 at 11:34 AM, Aljoscha Krettek <[hidden email]> wrote:
>> Maybe we can get rid of the serializer factories altogether. We could
>> enhance the Serializers with a method duplicate() that does nothing
>> for stateless serializers and does a deep copy for stateful
>> serializers. This would also consolidate all the knowledge about
>> stateful/stateless in one place.
>>
>> On Fri, Jan 30, 2015 at 11:25 AM, Aljoscha Krettek <[hidden email]> wrote:
>>> Of course it doesn't work. The ClassLoader is declared transient in
>>> the serializer factory, so it is Null once the factory has been
>>> serialized/deserialized once. I open a Jira issue:
>>> https://issues.apache.org/jira/browse/FLINK-1463
>>>
>>> On Thu, Jan 29, 2015 at 9:43 PM, Stephan Ewen <[hidden email]> wrote:
>>>> Thanks for reporting this, Arvid. I remember that we saw something similar
>>>> a while back. I will take a look at this later today (I am in PST zone
>>>> now), can hopefully fix this.
>>>>
>>>> Greetings,
>>>> Stephan

[Arvid Heise-2]

Sorry building the 0.9 branch took longer than expected. I will follow that
up on Monday.

Alternatively I would be grateful for a 0.8.0 patch ;)

Best,

Arvid

On Fri, Jan 30, 2015 at 5:09 PM, Aljoscha Krettek <[hidden email]>
wrote:

> Hi Arvid,
> I have a fix that I hope fixes your problem:
> https://github.com/aljoscha/flink/tree/serializer-factories-fix
>
> Could you try building it and running your example?
>
> Cheers,
> Aljoscha
>
> On Fri, Jan 30, 2015 at 3:30 PM, Aljoscha Krettek <[hidden email]>
> wrote:
> > We have a bit of a divide in how we handle TypeSerializer and
> > TypeComparator: TypeSerializer does not handle duplication but relies
> > on outside code (RuntimeStatefulSerializerFactory) to perform the
> > duplication. TypeComparator does duplication itself. There is also the
> > RuntimeComparatorFactory. This, however, does not perform duplication
> > but simply hands out the same TypeComparator instance multiple times.
> > With comparators, the user of the comparator is responsible for
> > calling duplicate() on the comparator.
> >
> > I started work on a branch where I add a duplicate() method to
> > TypeSerializer that simply return itself for stateless serializers and
> > does a deep copy if it is stateful. The TypeSerializers no longer have
> > method isStateful() and I replaced the two serializer factories by one
> > factory that always returns a "duplicate" of the serializer it holds.
> >
> > I think we should consolidate the two approaches. So either let the
> > factories handle all the duplication or let the user of the
> > comparator/serializer always handle duplication. I think the former is
> > better, since the latter makes it very easy to forget to duplicate.
> >
> > On Fri, Jan 30, 2015 at 11:34 AM, Aljoscha Krettek <[hidden email]>
> wrote:
> >> Maybe we can get rid of the serializer factories altogether. We could
> >> enhance the Serializers with a method duplicate() that does nothing
> >> for stateless serializers and does a deep copy for stateful
> >> serializers. This would also consolidate all the knowledge about
> >> stateful/stateless in one place.
> >>
> >> On Fri, Jan 30, 2015 at 11:25 AM, Aljoscha Krettek <[hidden email]>
> wrote:
> >>> Of course it doesn't work. The ClassLoader is declared transient in
> >>> the serializer factory, so it is Null once the factory has been
> >>> serialized/deserialized once. I open a Jira issue:
> >>> https://issues.apache.org/jira/browse/FLINK-1463
> >>>
> >>> On Thu, Jan 29, 2015 at 9:43 PM, Stephan Ewen <[hidden email]>
> wrote:
> >>>> Thanks for reporting this, Arvid. I remember that we saw something
> similar
> >>>> a while back. I will take a look at this later today (I am in PST zone
> >>>> now), can hopefully fix this.
> >>>>
> >>>> Greetings,
> >>>> Stephan
>

[Arvid Heise-2]

OK, patch indeed worked for my workflow. Thank you very much.

Any idea when this patch will be in a non-snapshot?

Best,

Arvid

On Fri, Jan 30, 2015 at 6:04 PM, Arvid Heise <[hidden email]> wrote:

> Sorry building the 0.9 branch took longer than expected. I will follow
> that up on Monday.
>
> Alternatively I would be grateful for a 0.8.0 patch ;)
>
> Best,
>
> Arvid
>
> On Fri, Jan 30, 2015 at 5:09 PM, Aljoscha Krettek <[hidden email]>
> wrote:
>
>> Hi Arvid,
>> I have a fix that I hope fixes your problem:
>> https://github.com/aljoscha/flink/tree/serializer-factories-fix
>>
>> Could you try building it and running your example?
>>
>> Cheers,
>> Aljoscha
>>
>> On Fri, Jan 30, 2015 at 3:30 PM, Aljoscha Krettek <[hidden email]>
>> wrote:
>> > We have a bit of a divide in how we handle TypeSerializer and
>> > TypeComparator: TypeSerializer does not handle duplication but relies
>> > on outside code (RuntimeStatefulSerializerFactory) to perform the
>> > duplication. TypeComparator does duplication itself. There is also the
>> > RuntimeComparatorFactory. This, however, does not perform duplication
>> > but simply hands out the same TypeComparator instance multiple times.
>> > With comparators, the user of the comparator is responsible for
>> > calling duplicate() on the comparator.
>> >
>> > I started work on a branch where I add a duplicate() method to
>> > TypeSerializer that simply return itself for stateless serializers and
>> > does a deep copy if it is stateful. The TypeSerializers no longer have
>> > method isStateful() and I replaced the two serializer factories by one
>> > factory that always returns a "duplicate" of the serializer it holds.
>> >
>> > I think we should consolidate the two approaches. So either let the
>> > factories handle all the duplication or let the user of the
>> > comparator/serializer always handle duplication. I think the former is
>> > better, since the latter makes it very easy to forget to duplicate.
>> >
>> > On Fri, Jan 30, 2015 at 11:34 AM, Aljoscha Krettek <[hidden email]>
>> wrote:
>> >> Maybe we can get rid of the serializer factories altogether. We could
>> >> enhance the Serializers with a method duplicate() that does nothing
>> >> for stateless serializers and does a deep copy for stateful
>> >> serializers. This would also consolidate all the knowledge about
>> >> stateful/stateless in one place.
>> >>
>> >> On Fri, Jan 30, 2015 at 11:25 AM, Aljoscha Krettek <
>> [hidden email]> wrote:
>> >>> Of course it doesn't work. The ClassLoader is declared transient in
>> >>> the serializer factory, so it is Null once the factory has been
>> >>> serialized/deserialized once. I open a Jira issue:
>> >>> https://issues.apache.org/jira/browse/FLINK-1463
>> >>>
>> >>> On Thu, Jan 29, 2015 at 9:43 PM, Stephan Ewen <[hidden email]>
>> wrote:
>> >>>> Thanks for reporting this, Arvid. I remember that we saw something
>> similar
>> >>>> a while back. I will take a look at this later today (I am in PST
>> zone
>> >>>> now), can hopefully fix this.
>> >>>>
>> >>>> Greetings,
>> >>>> Stephan
>>
>
>

[Aljoscha Krettek-2]

Hi,
we have some bug fixes queued up. So a 0.8.1 bug fix release should be
expected in the upcoming weeks.

Cheers,
Aljoscha

On Mon, Feb 2, 2015 at 12:48 PM, Arvid Heise <[hidden email]> wrote:

> OK, patch indeed worked for my workflow. Thank you very much.
>
> Any idea when this patch will be in a non-snapshot?
>
> Best,
>
> Arvid
>
> On Fri, Jan 30, 2015 at 6:04 PM, Arvid Heise <[hidden email]> wrote:
>
>> Sorry building the 0.9 branch took longer than expected. I will follow
>> that up on Monday.
>>
>> Alternatively I would be grateful for a 0.8.0 patch ;)
>>
>> Best,
>>
>> Arvid
>>
>> On Fri, Jan 30, 2015 at 5:09 PM, Aljoscha Krettek <[hidden email]>
>> wrote:
>>
>>> Hi Arvid,
>>> I have a fix that I hope fixes your problem:
>>> https://github.com/aljoscha/flink/tree/serializer-factories-fix
>>>
>>> Could you try building it and running your example?
>>>
>>> Cheers,
>>> Aljoscha
>>>
>>> On Fri, Jan 30, 2015 at 3:30 PM, Aljoscha Krettek <[hidden email]>
>>> wrote:
>>> > We have a bit of a divide in how we handle TypeSerializer and
>>> > TypeComparator: TypeSerializer does not handle duplication but relies
>>> > on outside code (RuntimeStatefulSerializerFactory) to perform the
>>> > duplication. TypeComparator does duplication itself. There is also the
>>> > RuntimeComparatorFactory. This, however, does not perform duplication
>>> > but simply hands out the same TypeComparator instance multiple times.
>>> > With comparators, the user of the comparator is responsible for
>>> > calling duplicate() on the comparator.
>>> >
>>> > I started work on a branch where I add a duplicate() method to
>>> > TypeSerializer that simply return itself for stateless serializers and
>>> > does a deep copy if it is stateful. The TypeSerializers no longer have
>>> > method isStateful() and I replaced the two serializer factories by one
>>> > factory that always returns a "duplicate" of the serializer it holds.
>>> >
>>> > I think we should consolidate the two approaches. So either let the
>>> > factories handle all the duplication or let the user of the
>>> > comparator/serializer always handle duplication. I think the former is
>>> > better, since the latter makes it very easy to forget to duplicate.
>>> >
>>> > On Fri, Jan 30, 2015 at 11:34 AM, Aljoscha Krettek <[hidden email]>
>>> wrote:
>>> >> Maybe we can get rid of the serializer factories altogether. We could
>>> >> enhance the Serializers with a method duplicate() that does nothing
>>> >> for stateless serializers and does a deep copy for stateful
>>> >> serializers. This would also consolidate all the knowledge about
>>> >> stateful/stateless in one place.
>>> >>
>>> >> On Fri, Jan 30, 2015 at 11:25 AM, Aljoscha Krettek <
>>> [hidden email]> wrote:
>>> >>> Of course it doesn't work. The ClassLoader is declared transient in
>>> >>> the serializer factory, so it is Null once the factory has been
>>> >>> serialized/deserialized once. I open a Jira issue:
>>> >>> https://issues.apache.org/jira/browse/FLINK-1463
>>> >>>
>>> >>> On Thu, Jan 29, 2015 at 9:43 PM, Stephan Ewen <[hidden email]>
>>> wrote:
>>> >>>> Thanks for reporting this, Arvid. I remember that we saw something
>>> similar
>>> >>>> a while back. I will take a look at this later today (I am in PST
>>> zone
>>> >>>> now), can hopefully fix this.
>>> >>>>
>>> >>>> Greetings,
>>> >>>> Stephan
>>>
>>
>>