enableObjectReuse and multiple downstream operators

Posted by Bart Wyatt on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/enableObjectReuse-and-multiple-downstream-operators-tp7176.html

(For reference, I'm in 1.0.3) 

I have a job that looks like this:

DataStream<A> input = ...

input

    .map(MapFunction<A,B>...)

    .addSink(...);

input

    .map(MapFunction<A,C>...)

    ​.addSink(...);

If I do not call enableObjectReuse() it works, if I do call enableObjectReuse() it throws:

java.lang.ClassCastException: B cannot be cast to A

when attempting to process the second MapFunction<A,C>

It looks like in the input operator is calling Output<StreamRecord<A>>.collect, which will pass the value to the first MapFunction and then the second MapFunction.  However, the first map function calls StreamRecord<>.replace() which mutates the stored value to the output of the function.  When Output<StreamRecord<A>>.collect passes this to the next MapFunction it is now as if the two MapFunctions are in serial and not parallel.

I looked into JIRA and didn't see an issue that looked exactly like this.  Is this known?

-Bart

---

This e-mail may contain CONFIDENTIAL AND PROPRIETARY INFORMATION and/or PRIVILEGED AND CONFIDENTIAL COMMUNICATION intended solely for the recipient and, therefore, may not be retransmitted to any party outside of the recipient's organization without the prior written consent of the sender. If you have received this e-mail in error please notify the sender immediately by telephone or reply e-mail and destroy the original message without making a copy. Deep Silver, Inc. accepts no liability for any losses or damages resulting from infected e-mail transmissions and viruses in e-mail attachments.