[FLINK-12113] User code passing to fromCollection(Iterator, Class) not cleaned

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.7.2
• Component/s: API / DataStream
• Labels: auto-deprioritized-major auto-deprioritized-minor auto-unassigned
• Assignee: Unassigned
• Reporter: yankai zhang
• Created: 04/Apr/19 16:16
• Updated: 14/Mar/22 10:42

Description

 

interface IS<E> extends Iterator<E>, Serializable { }

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.fromCollection(new IS<Object>() {
    @Override
    public boolean hasNext() {
        return false;
    }

    @Override
    public Object next() {
        return null;
    }
}, Object.class);

Code piece above throws exception:

org.apache.flink.api.common.InvalidProgramException: The implementation of the SourceFunction is not serializable. The object probably contains or references non serializable fields.

  at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:99)
....

And my workaround is wrapping clean around iterator instance, like this:

 

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.fromCollection(env.clean(new IS<Object>() {
    @Override
    public boolean hasNext() {
        return false;
    }

    @Override
    public Object next() {
        return null;
    }
}), Object.class);