@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));
  }
@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)));
  }