TengHu commented on a change in pull request #8885:
URL: https://github.com/apache/flink/pull/8885#discussion_r429491587
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/assigners/TumblingProcessingTimeWindows.java
##########
@@ -46,28 +46,44 @@
private final long size;
- private final long offset;
+ private final long globalOffset;
- private TumblingProcessingTimeWindows(long size, long offset) {
+ private Long staggerOffset = null;
+
+ private final WindowStagger windowStagger;
+
+ private TumblingProcessingTimeWindows(long size, long offset,
WindowStagger windowStagger) {
if (Math.abs(offset) >= size) {
throw new
IllegalArgumentException("TumblingProcessingTimeWindows parameters must satisfy
abs(offset) < size");
}
-
this.size = size;
- this.offset = offset;
+ this.globalOffset = offset;
+ this.windowStagger = windowStagger;
}
@Override
public Collection<TimeWindow> assignWindows(Object element, long
timestamp, WindowAssignerContext context) {
final long now = context.getCurrentProcessingTime();
- long start = TimeWindow.getWindowStartWithOffset(now, offset,
size);
+
+ if (staggerOffset == null) {
+ staggerOffset =
windowStagger.getStaggerOffset(context.getCurrentProcessingTime(), size);
Review comment:
The intention was to align the pane to its first event. From our
experience, the offsets among partitions were usually caused by the design of
partition keys, network delay, etc, which led to a normal distribution
staggering.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
