WindmillKeyedWorkItem implements KeyedWorkItem<K, ElemT> { @Override public Iterable<TimerData> timersIterable() { FluentIterable<Timer> allTimers = FluentIterable.from(workItem.getTimers().getTimersList()); FluentIterable<Timer> eventTimers = allTimers.filter(IS_WATERMARK); FluentIterable<Timer> nonEventTimers = allTimers.filter(Predicates.not(IS_WATERMARK)); return eventTimers .append(nonEventTimers) .transform( timer -> WindmillTimerInternals.windmillTimerToTimerData( WindmillNamespacePrefix.SYSTEM_NAMESPACE_PREFIX, timer, windowCoder)); } WindmillKeyedWorkItem(
      K key,
      Windmill.WorkItem workItem,
      Coder<? extends BoundedWindow> windowCoder,
      Coder<Collection<? extends BoundedWindow>> windowsCoder,
      Coder<ElemT> valueCoder); @Override K key(); @Override Iterable<TimerData> timersIterable(); @Override Iterable<WindowedValue<ElemT>> elementsIterable(); @Override boolean equals(@Nullable Object other); @Override int hashCode(); @Override String toString();  }
@Test public void testTimerOrdering() throws Exception { Windmill.WorkItem workItem = Windmill.WorkItem.newBuilder() .setKey(SERIALIZED_KEY) .setWorkToken(17) .setTimers( Windmill.TimerBundle.newBuilder() .addTimers( makeSerializedTimer(STATE_NAMESPACE_1, 0, Windmill.Timer.Type.REALTIME)) .addTimers( makeSerializedTimer(STATE_NAMESPACE_1, 1, Windmill.Timer.Type.WATERMARK)) .addTimers( makeSerializedTimer(STATE_NAMESPACE_1, 2, Windmill.Timer.Type.REALTIME)) .addTimers( makeSerializedTimer(STATE_NAMESPACE_2, 3, Windmill.Timer.Type.WATERMARK)) .build()) .build(); KeyedWorkItem<String, String> keyedWorkItem = new WindmillKeyedWorkItem<>(KEY, workItem, WINDOW_CODER, WINDOWS_CODER, VALUE_CODER); assertThat( keyedWorkItem.timersIterable(), Matchers.contains( makeTimer(STATE_NAMESPACE_1, 1, TimeDomain.EVENT_TIME), makeTimer(STATE_NAMESPACE_2, 3, TimeDomain.EVENT_TIME), makeTimer(STATE_NAMESPACE_1, 0, TimeDomain.PROCESSING_TIME), makeTimer(STATE_NAMESPACE_1, 2, TimeDomain.PROCESSING_TIME))); }