[hotfix] Properly copy stream record in ReducingWindowBuffer and FoldingWindowBuffer

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/buffers/FoldingWindowBuffer.java

@@ -63,7 +63,7 @@ public class FoldingWindowBuffer<T, ACC> implements WindowBuffer<T, ACC> {
 
 	@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

flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/buffers/ReducingWindowBuffer.java

@@ -57,7 +57,7 @@ public class ReducingWindowBuffer<T> implements WindowBuffer<T, 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()));
 		}