You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2019/03/11 17:21:19 UTC
[beam] branch master updated: Update ReshuffleOverrideFactory to
keep Reshuffle's TimestampCombiner.EARLIEST setting
This is an automated email from the ASF dual-hosted git repository.
kenn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 57183b8 Update ReshuffleOverrideFactory to keep Reshuffle's TimestampCombiner.EARLIEST setting
new 1dafb54 Merge pull request #8025: Update ReshuffleOverrideFactory to keep Reshuffle's TimestampCombiner.EARLIEST setting
57183b8 is described below
commit 57183b834b2dacfc552c8ed768b426ae7cc5b3dc
Author: CraigChambersG <45...@users.noreply.github.com>
AuthorDate: Sun Mar 10 10:41:14 2019 -0700
Update ReshuffleOverrideFactory to keep Reshuffle's TimestampCombiner.EARLIEST setting
R: @kennknowles
R: @robertwb
---
.../java/org/apache/beam/runners/dataflow/ReshuffleOverrideFactory.java | 2 ++
1 file changed, 2 insertions(+)
diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/ReshuffleOverrideFactory.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/ReshuffleOverrideFactory.java
index 0ee620a..17f92bc 100644
--- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/ReshuffleOverrideFactory.java
+++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/ReshuffleOverrideFactory.java
@@ -28,6 +28,7 @@ import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.Reshuffle;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.ReshuffleTrigger;
+import org.apache.beam.sdk.transforms.windowing.TimestampCombiner;
import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.util.IdentityWindowFn;
import org.apache.beam.sdk.values.KV;
@@ -65,6 +66,7 @@ class ReshuffleOverrideFactory<K, V>
new IdentityWindowFn<>(originalStrategy.getWindowFn().windowCoder()))
.triggering(new ReshuffleTrigger<>())
.discardingFiredPanes()
+ .withTimestampCombiner(TimestampCombiner.EARLIEST)
.withAllowedLateness(Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()));
return input