aljoscha commented on a change in pull request #12297:
URL: https://github.com/apache/flink/pull/12297#discussion_r429763639
##########
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:
I think my comment from your other PR still hold: Does
WindowStagger.NATURAL make sense if we keep the stagger offset fixed per
WindowAssigner instance? I don't know if this was the intention but that's
what
the code does.
This will set a fixed offset when the first event is received, Flink will
not create a new `WindowAssigner` instance for new panes, so those new panes
will use the same offset that was set for the first pane that we ever saw.
That's why I'm saying that `NATURAL` doesn't really do anything here. (Or I'm
misunderstanding sth.)
----------------------------------------------------------------
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:
us...@infra.apache.org
