flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Sebastian Neef <gehax...@mailbox.tu-berlin.de>
Subject Weird serialization bug?
Date Sat, 29 Apr 2017 11:26:58 GMT
Hello Apache Flink users,

I implemented a FilterFunction some months ago that worked quite well
back then. However, I wanted to check it out right now and it somehow
broke in the sense that Flink can't serialize it anymore. I might be
mistaken, but afaik I didn't touch the code at all.

I think that I've tracked down the problem to the following minimal
working PoC:

- A simple interface:

>     interface testFunc extends Serializable {
>         boolean filter();
>     }

- A TestFilterFunction which is applied on a DataSet:

>  public void doSomeFiltering() {
>         class fooo implements testFunc {
>             public boolean filter() {
>                 return false;
>             }
>         }
> 
>         class TestFilterFunction implements FilterFunction<IPage> {
> 
>             testFunc filter;
> 
>             class fooo2 implements testFunc {
>                 public boolean filter() {
>                     return false;
>                 }
>             }
> 
>             TestFilterFunction() {
>                 // WORKS with fooo2()
>                 // DOES NOT WORK with fooo()
>                 this.filter = new fooo2();
>             }
>             @Override
>             public boolean filter(IPage iPage) throws Exception {
>                 return filter.filter();
>             }
>         }
> 	filteredDataSet = DataSet.filter(new TestFilterFunction(null))> }

Flink will work fine when the "fooo2" class is used. However, when using
the "fooo()" class, I get the following error:

> ------------------------------------------------------------
>  The program finished with the following exception:
> 
> The implementation of the FilterFunction is not serializable. The object probably contains
or references non serializable fields.
> 	org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:100)
> 	org.apache.flink.api.java.DataSet.clean(DataSet.java:186)
> 	org.apache.flink.api.java.DataSet.filter(DataSet.java:287)
> 	testpackage.testclass.applyFilters(testclass.java:105)

I'm a little bit confused, why Flink manages to serialize the "fooo2"
class, but not the "fooo" class. Is this is a bug or do I miss something
here?

Cheers,
Sebastian


Mime
View raw message