Github user ggevay commented on a diff in the pull request:
https://github.com/apache/flink/pull/1517#discussion_r65231386
--- Diff:
flink-runtime/src/main/java/org/apache/flink/runtime/operators/ReduceCombineDriver.java
---
@@ -42,34 +44,38 @@
* Combine operator for Reduce functions, standalone (not chained).
* Sorts and groups and reduces data, but never spills the sort. May
produce multiple
* partially aggregated groups.
- *
+ *
* @param <T> The data type consumed and produced by the combiner.
*/
public class ReduceCombineDriver<T> implements Driver<ReduceFunction<T>,
T> {
-
+
private static final Logger LOG =
LoggerFactory.getLogger(ReduceCombineDriver.class);
/** Fix length records with a length below this threshold will be
in-place sorted, if possible. */
private static final int THRESHOLD_FOR_IN_PLACE_SORTING = 32;
-
-
+
+
private TaskContext<ReduceFunction<T>, T> taskContext;
private TypeSerializer<T> serializer;
private TypeComparator<T> comparator;
-
+
private ReduceFunction<T> reducer;
-
+
private Collector<T> output;
-
+
+ private DriverStrategy strategy;
+
private InMemorySorter<T> sorter;
-
+
private QuickSort sortAlgo = new QuickSort();
+ private ReduceHashTable<T> table;
+
private List<MemorySegment> memory;
- private boolean running;
+ private volatile boolean canceled;
--- End diff --
> Can we keep the name of the flag as it is for now? All other drivers use
a running flag as well.
I would rather open a separate JIRA to fix the name in all drivers.
12e36ab93b7e7d94d497a6d718eba21ead813d7e
And here is the Jira: https://issues.apache.org/jira/browse/FLINK-3999
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
