[FLINK-6394] GroupCombine reuses instances even though object reuse is disabled

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.2.0
• Fix Version/s: 1.2.2, 1.3.0
• Component/s: API / DataSet
• Assignee: Kurt Young
• Reporter: Jaromir Vanek
• Created: 27/Apr/17 11:45
• Updated: 08/May/17 12:24
• Resolved: 08/May/17 12:24

Description

I am using group combiner in DataSet API with disabled object reuse.

In code it may be expressed as follows:

tuples.groupBy(1)
      .combineGroup((it, collector) -> {
         // store first item for future use
         Pojo stored = it.next();
         while (it.hasNext()) {
           ....
         }
      })

It seems even the object reuse feature is disabled, my instance is actually replaced when .next() is called on the iterator. It leads to very confusing and wrong results.

I checked the Flink codebase and it seems CombiningUnilateralSortMerger is actually reusing object instances even though object reuse is explicitly disabled.
In spilling phase user's combiner is called with instance of CombineValueIterator that actually reuses instances without any warning.

See https://github.com/apache/flink/blob/d7b59d761601baba6765bb4fc407bcd9fd6a9387/flink-runtime/src/main/java/org/apache/flink/runtime/operators/sort/CombiningUnilateralSortMerger.java#L550

When I disable combiner and use groupReduce only with the same reduce function, results are fine.

Please let me know if you can confirm this as a bug. From my point of view it's highly critical as I am getting unpredictable results.

Issue Links

• links to: GitHub Pull Request #3803