aljoscha commented on a change in pull request #12297:
URL: https://github.com/apache/flink/pull/12297#discussion_r436527795
##########
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:
Ok, now I understand. Thanks!
----------------------------------------------------------------
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]
