WindowOperatorSpec extends OperatorSpec<M, WindowPane<WK, WV>> implements StatefulOperatorSpec { public long getDefaultTriggerMs() { List<TimeBasedTrigger> timeBasedTriggers = new ArrayList<>(); if (window.getDefaultTrigger() != null) { timeBasedTriggers.addAll(getTimeBasedTriggers(window.getDefaultTrigger())); } if (window.getEarlyTrigger() != null) { timeBasedTriggers.addAll(getTimeBasedTriggers(window.getEarlyTrigger())); } if (window.getLateTrigger() != null) { timeBasedTriggers.addAll(getTimeBasedTriggers(window.getLateTrigger())); } LOG.info("Got {} time-based triggers", timeBasedTriggers.size()); List<Long> candidateDurations = timeBasedTriggers.stream() .map(timeBasedTrigger -> timeBasedTrigger.getDuration().toMillis()) .collect(Collectors.toList()); return MathUtil.gcd(candidateDurations); } WindowOperatorSpec(WindowInternal<M, WK, WV> window, String opId); WindowInternal<M, WK, WV> getWindow(); long getDefaultTriggerMs(); @Override WatermarkFunction getWatermarkFn(); @Override ScheduledFunction getScheduledFn(); @Override Collection<StoreDescriptor> getStoreDescriptors();  }
@Test public void testTriggerIntervalWithNestedTimeTriggers() { defaultTrigger = Triggers.timeSinceFirstMessage(Duration.ofMillis(150)); lateTrigger = Triggers.any(Triggers.count(6), Triggers.timeSinceFirstMessage(Duration.ofMillis(15))); earlyTrigger = Triggers.repeat( Triggers.any(Triggers.count(23), Triggers.timeSinceFirstMessage(Duration.ofMillis(15)), Triggers.any(Triggers.any(Triggers.count(6), Triggers.timeSinceFirstMessage(Duration.ofMillis(15)), Triggers.timeSinceFirstMessage(Duration.ofMillis(25)), Triggers.timeSinceLastMessage(Duration.ofMillis(15)))))); WindowOperatorSpec spec = getWindowOperatorSpec("w0"); assertEquals(spec.getDefaultTriggerMs(), 5); }
@Test public void testTriggerIntervalWithSingleTimeTrigger() { WindowOperatorSpec spec = getWindowOperatorSpec("w0"); assertEquals(spec.getDefaultTriggerMs(), 150); }