[hotfix] Properly copy stream record in ReducingWindowBuffer and Fold…

…ingWindowBuffer
tchanghunua · Feb 29, 2016 · 405d222 · 405d222
1 parent f881e70
commit 405d222
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/...a/org/apache/flink/streaming/runtime/operators/windowing/buffers/FoldingWindowBuffer.java b/...a/org/apache/flink/streaming/runtime/operators/windowing/buffers/FoldingWindowBuffer.java
@@ -63,7 +63,7 @@ protected FoldingWindowBuffer(FoldFunction<T, ACC> foldFunction, StreamRecord<AC
 
  @Override
  public void storeElement(StreamRecord<T> element) throws Exception {
- data.replace(foldFunction.fold(data.getValue(), element.getValue()), element.getTimestamp());
+ data.replace(foldFunction.fold(data.getValue(), element.getValue()));
  }
 
  @Override

diff --git a/.../org/apache/flink/streaming/runtime/operators/windowing/buffers/ReducingWindowBuffer.java b/.../org/apache/flink/streaming/runtime/operators/windowing/buffers/ReducingWindowBuffer.java
@@ -57,7 +57,7 @@ protected ReducingWindowBuffer(ReduceFunction<T> reduceFunction, StreamRecord<T>
  @Override
  public void storeElement(StreamRecord<T> element) throws Exception {
  if (data == null) {
- data = new StreamRecord<>(element.getValue(), element.getTimestamp());
+ data = element.copy(element.getValue());
  } else {
  data.replace(reduceFunction.reduce(data.getValue(), element.getValue()));
  }