/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.beam.runners.direct;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.contains;
import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.emptyIterable;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.lessThan;
import static org.hamcrest.Matchers.not;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.when;

import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.beam.runners.core.StateNamespaces;
import org.apache.beam.runners.core.TimerInternals.TimerData;
import org.apache.beam.runners.direct.WatermarkManager.AppliedPTransformInputWatermark;
import org.apache.beam.runners.direct.WatermarkManager.FiredTimers;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate;
import org.apache.beam.runners.direct.WatermarkManager.TimerUpdate.TimerUpdateBuilder;
import org.apache.beam.runners.direct.WatermarkManager.TransformWatermarks;
import org.apache.beam.runners.direct.WatermarkManager.Watermark;
import org.apache.beam.runners.local.StructuralKey;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.runners.AppliedPTransform;
import org.apache.beam.sdk.state.TimeDomain;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.transforms.Flatten;
import org.apache.beam.sdk.transforms.Impulse;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.WithKeys;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
import org.apache.beam.sdk.transforms.windowing.IntervalWindow;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
import org.apache.beam.sdk.util.WindowedValue;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionList;
import org.apache.beam.sdk.values.TimestampedValue;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
import org.joda.time.Duration;
import org.joda.time.Instant;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.Mockito;

/**
 * Tests for {@link WatermarkManager}.
 *
 * <p>Note that the tests below output multiple elements from impulse for the purpose of watermark
 * tracking while impulse would normally only output a single empty byte array in the global window.
 */
@RunWith(JUnit4.class)
@SuppressWarnings({
  "rawtypes", // TODO(https://github.com/apache/beam/issues/20447)
})
public class WatermarkManagerTest implements Serializable {
  @Rule public transient ExpectedException thrown = ExpectedException.none();

  private transient MockClock clock;

  private transient PCollection<byte[]> impulse;

  private transient PCollection<byte[]> filtered;
  private transient PCollection<byte[]> filteredNotEmpty;
  private transient PCollection<KV<String, byte[]>> keyed;

  private transient PCollection<byte[]> impulseToFlatten;
  private transient PCollection<byte[]> flattened;

  private transient WatermarkManager<AppliedPTransform<?, ?, ?>, ? super PCollection<?>> manager;
  private transient BundleFactory bundleFactory;
  private DirectGraph graph;

  @Rule
  public transient TestPipeline p = TestPipeline.create().enableAbandonedNodeEnforcement(false);

  @Before
  public void setup() {

    impulse = p.apply(Impulse.create());
    filtered = impulse.apply("filtered", Filter.by((element) -> element.length == 0));
    filteredNotEmpty =
        filtered.apply(
            "timesTwo",
            ParDo.of(
                new DoFn<byte[], byte[]>() {
                  @ProcessElement
                  public void processElement(ProcessContext c) throws Exception {
                    if (c.element().length > 0) {
                      c.output(c.element());
                    }
                  }
                }));

    keyed = impulse.apply("keyed", WithKeys.of("MyKey"));

    impulseToFlatten = p.apply("impulseToFlatten", Impulse.create());
    PCollectionList<byte[]> preFlatten = PCollectionList.of(impulse).and(impulseToFlatten);
    flattened = preFlatten.apply("flattened", Flatten.pCollections());

    clock = MockClock.fromInstant(new Instant(1000));
    DirectGraphs.performDirectOverrides(p);
    graph = DirectGraphs.getGraph(p);

    manager = WatermarkManager.create(clock, graph, AppliedPTransform::getFullName);
    bundleFactory = ImmutableListBundleFactory.create();
  }

  /**
   * Demonstrates that getWatermark, when called on an {@link AppliedPTransform} that has not
   * processed any elements, returns the {@link BoundedWindow#TIMESTAMP_MIN_VALUE}.
   */
  @Test
  public void getWatermarkForUntouchedTransform() {
    TransformWatermarks watermarks = manager.getWatermarks(graph.getProducer(impulse));

    assertThat(watermarks.getInputWatermark(), equalTo(BoundedWindow.TIMESTAMP_MIN_VALUE));
    assertThat(watermarks.getOutputWatermark(), equalTo(BoundedWindow.TIMESTAMP_MIN_VALUE));
  }

  /**
   * Demonstrates that getWatermark for a transform that consumes no input uses the Watermark Hold
   * value provided to it as the output watermark.
   */
  @Test
  public void getWatermarkForUpdatedSourceTransform() {
    CommittedBundle<byte[]> output = multiWindowedBundle(impulse, new byte[1]);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.singleton(output),
        new Instant(8000L));
    manager.refreshAll();
    TransformWatermarks updatedSourceWatermark = manager.getWatermarks(graph.getProducer(impulse));

    assertThat(updatedSourceWatermark.getOutputWatermark(), equalTo(new Instant(8000L)));
  }

  /**
   * Demonstrates that getWatermark for a transform that takes multiple inputs is held to the
   * minimum watermark across all of its inputs.
   */
  @Test
  public void getWatermarkForMultiInputTransform() {
    CommittedBundle<byte[]> secondPcollectionBundle =
        multiWindowedBundle(impulseToFlatten, new byte[1]);

    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulseToFlatten),
        null,
        Collections.<CommittedBundle<?>>singleton(secondPcollectionBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();

    // We didn't do anything for the first source, so we shouldn't have progressed the watermark
    TransformWatermarks firstSourceWatermark = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(
        firstSourceWatermark.getOutputWatermark(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));

    // the Second Source output all of the elements so it should be done (with a watermark at the
    // end of time).
    TransformWatermarks secondSourceWatermark =
        manager.getWatermarks(graph.getProducer(impulseToFlatten));
    assertThat(
        secondSourceWatermark.getOutputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));

    // We haven't consumed anything yet, so our watermark should be at the beginning of time
    TransformWatermarks transformWatermark = manager.getWatermarks(graph.getProducer(flattened));
    assertThat(
        transformWatermark.getInputWatermark(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));
    assertThat(
        transformWatermark.getOutputWatermark(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));

    CommittedBundle<byte[]> flattenedBundleSecondCreate =
        multiWindowedBundle(flattened, new byte[1]);
    // We have finished processing the bundle from the second PCollection, but we haven't consumed
    // anything from the first PCollection yet; so our watermark shouldn't advance
    manager.updateWatermarks(
        secondPcollectionBundle,
        TimerUpdate.empty(),
        graph.getProducer(flattened),
        secondPcollectionBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(flattenedBundleSecondCreate),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    TransformWatermarks transformAfterProcessing =
        manager.getWatermarks(graph.getProducer(flattened));
    manager.updateWatermarks(
        secondPcollectionBundle,
        TimerUpdate.empty(),
        graph.getProducer(flattened),
        secondPcollectionBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(flattenedBundleSecondCreate),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    assertThat(
        transformAfterProcessing.getInputWatermark(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));
    assertThat(
        transformAfterProcessing.getOutputWatermark(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));

    Instant firstCollectionTimestamp = new Instant(10000);
    CommittedBundle<byte[]> firstPcollectionBundle =
        timestampedBundle(impulse, TimestampedValue.of(new byte[5], firstCollectionTimestamp));
    // the source is done, but elements are still buffered. The source output watermark should be
    // past the end of the global window
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(firstPcollectionBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks firstSourceWatermarks = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(
        firstSourceWatermarks.getOutputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));

    // We still haven't consumed any of the first source's input, so the watermark should still not
    // progress
    TransformWatermarks flattenAfterSourcesProduced =
        manager.getWatermarks(graph.getProducer(flattened));
    assertThat(
        flattenAfterSourcesProduced.getInputWatermark(),
        not(greaterThan(firstCollectionTimestamp)));
    assertThat(
        flattenAfterSourcesProduced.getOutputWatermark(),
        not(greaterThan(firstCollectionTimestamp)));

    // We have buffered inputs, but since the PCollection has all of the elements (has a WM past the
    // end of the global window), we should have a watermark equal to the min among buffered
    // elements
    TransformWatermarks withBufferedElements = manager.getWatermarks(graph.getProducer(flattened));
    assertThat(withBufferedElements.getInputWatermark(), equalTo(firstCollectionTimestamp));
    assertThat(withBufferedElements.getOutputWatermark(), equalTo(firstCollectionTimestamp));

    CommittedBundle<?> completedFlattenBundle =
        bundleFactory.createBundle(flattened).commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.updateWatermarks(
        firstPcollectionBundle,
        TimerUpdate.empty(),
        graph.getProducer(flattened),
        firstPcollectionBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(completedFlattenBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks afterConsumingAllInput =
        manager.getWatermarks(graph.getProducer(flattened));
    assertThat(
        afterConsumingAllInput.getInputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    assertThat(
        afterConsumingAllInput.getOutputWatermark(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
  }

  /**
   * Demonstrates that getWatermark for a transform that takes multiple inputs is held to the
   * minimum watermark across all of its inputs.
   */
  @Test
  public void getWatermarkMultiIdenticalInput() {
    PCollection<byte[]> impulse = p.apply(Impulse.create());
    PCollection<byte[]> multiConsumer =
        PCollectionList.of(impulse).and(impulse).apply(Flatten.pCollections());
    DirectGraphVisitor graphVisitor = new DirectGraphVisitor();
    p.traverseTopologically(graphVisitor);
    DirectGraph graph = graphVisitor.getGraph();

    AppliedPTransform<?, ?, ?> theFlatten = graph.getProducer(multiConsumer);

    WatermarkManager<AppliedPTransform<?, ?, ?>, ? super PCollection<?>> tstMgr =
        WatermarkManager.create(clock, graph, AppliedPTransform::getFullName);
    CommittedBundle<Void> root =
        bundleFactory
            .<Void>createRootBundle()
            .add(WindowedValue.valueInGlobalWindow(null))
            .commit(clock.now());
    CommittedBundle<byte[]> createBundle =
        bundleFactory
            .createBundle(impulse)
            .add(WindowedValue.timestampedValueInGlobalWindow(new byte[1], new Instant(33536)))
            .commit(clock.now());

    Map<AppliedPTransform<?, ?, ?>, Collection<CommittedBundle<?>>> initialInputs =
        ImmutableMap.<AppliedPTransform<?, ?, ?>, Collection<CommittedBundle<?>>>builder()
            .put(graph.getProducer(impulse), Collections.singleton(root))
            .build();
    tstMgr.initialize((Map) initialInputs);
    tstMgr.updateWatermarks(
        root,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.singleton(createBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);

    tstMgr.refreshAll();
    TransformWatermarks flattenWms = tstMgr.getWatermarks(theFlatten);
    assertThat(flattenWms.getInputWatermark(), equalTo(new Instant(33536)));

    tstMgr.updateWatermarks(
        createBundle,
        TimerUpdate.empty(),
        theFlatten,
        null,
        Collections.emptyList(),
        BoundedWindow.TIMESTAMP_MAX_VALUE);

    tstMgr.refreshAll();
    assertThat(flattenWms.getInputWatermark(), equalTo(new Instant(33536)));

    tstMgr.updateWatermarks(
        createBundle,
        TimerUpdate.empty(),
        theFlatten,
        null,
        Collections.emptyList(),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    tstMgr.refreshAll();
    assertThat(flattenWms.getInputWatermark(), equalTo(BoundedWindow.TIMESTAMP_MAX_VALUE));
  }

  /**
   * Demonstrates that pending elements are independent among {@link AppliedPTransform
   * AppliedPTransforms} that consume the same input {@link PCollection}.
   */
  @Test
  public void getWatermarkForMultiConsumedCollection() {
    CommittedBundle<byte[]> impulseBundle =
        timestampedBundle(
            impulse,
            TimestampedValue.of(new byte[1], new Instant(1_000_000L)),
            TimestampedValue.of(new byte[2], new Instant(1234L)),
            TimestampedValue.of(new byte[3], new Instant(-1000L)));
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        new Instant(Long.MAX_VALUE));
    manager.refreshAll();
    TransformWatermarks impulseAfterProducing = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(
        impulseAfterProducing.getOutputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));

    CommittedBundle<KV<String, byte[]>> keyBundle =
        timestampedBundle(
            keyed,
            TimestampedValue.of(KV.of("MyKey", new byte[1]), new Instant(1_000_000L)),
            TimestampedValue.of(KV.of("MyKey", new byte[2]), new Instant(1234L)),
            TimestampedValue.of(KV.of("MyKey", new byte[3]), new Instant(-1000L)));
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(keyed),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(keyBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks keyedWatermarks = manager.getWatermarks(graph.getProducer(keyed));
    assertThat(
        keyedWatermarks.getInputWatermark(), not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    assertThat(
        keyedWatermarks.getOutputWatermark(), not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));

    TransformWatermarks filteredWatermarks = manager.getWatermarks(graph.getProducer(filtered));
    assertThat(filteredWatermarks.getInputWatermark(), not(greaterThan(new Instant(-1000L))));
    assertThat(filteredWatermarks.getOutputWatermark(), not(greaterThan(new Instant(-1000L))));

    CommittedBundle<byte[]> filteredBundle =
        timestampedBundle(filtered, TimestampedValue.of(new byte[2], new Instant(1234L)));
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(filteredBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks filteredProcessedWatermarks =
        manager.getWatermarks(graph.getProducer(filtered));
    assertThat(
        filteredProcessedWatermarks.getInputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    assertThat(
        filteredProcessedWatermarks.getOutputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
  }

  /**
   * Demonstrates that the watermark of an {@link AppliedPTransform} is held to the provided
   * watermark hold.
   */
  @Test
  public void updateWatermarkWithWatermarkHolds() {
    CommittedBundle<byte[]> impulseBundle =
        timestampedBundle(
            impulse,
            TimestampedValue.of(new byte[1], new Instant(1_000_000L)),
            TimestampedValue.of(new byte[2], new Instant(1234L)),
            TimestampedValue.of(new byte[3], new Instant(-1000L)));
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        new Instant(Long.MAX_VALUE));

    CommittedBundle<KV<String, byte[]>> keyBundle =
        timestampedBundle(
            keyed,
            TimestampedValue.of(KV.of("MyKey", new byte[1]), new Instant(1_000_000L)),
            TimestampedValue.of(KV.of("MyKey", new byte[2]), new Instant(1234L)),
            TimestampedValue.of(KV.of("MyKey", new byte[3]), new Instant(-1000L)));
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(keyed),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(keyBundle),
        new Instant(500L));
    manager.refreshAll();
    TransformWatermarks keyedWatermarks = manager.getWatermarks(graph.getProducer(keyed));
    assertThat(
        keyedWatermarks.getInputWatermark(), not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    assertThat(keyedWatermarks.getOutputWatermark(), not(greaterThan(new Instant(500L))));
  }

  /**
   * Demonstrates that the watermark of an {@link AppliedPTransform} is held to the provided
   * watermark hold.
   */
  @Test
  public void updateWatermarkWithKeyedWatermarkHolds() {
    CommittedBundle<byte[]> firstKeyBundle =
        bundleFactory
            .createKeyedBundle(StructuralKey.of("Odd", StringUtf8Coder.of()), impulse)
            .add(WindowedValue.timestampedValueInGlobalWindow(new byte[1], new Instant(1_000_000L)))
            .add(WindowedValue.timestampedValueInGlobalWindow(new byte[3], new Instant(-1000L)))
            .commit(clock.now());

    CommittedBundle<byte[]> secondKeyBundle =
        bundleFactory
            .createKeyedBundle(StructuralKey.of("Even", StringUtf8Coder.of()), impulse)
            .add(WindowedValue.timestampedValueInGlobalWindow(new byte[2], new Instant(1234L)))
            .commit(clock.now());

    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        ImmutableList.of(firstKeyBundle, secondKeyBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);

    manager.updateWatermarks(
        firstKeyBundle,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        firstKeyBundle.withElements(Collections.emptyList()),
        Collections.emptyList(),
        new Instant(-1000L));
    manager.updateWatermarks(
        secondKeyBundle,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        secondKeyBundle.withElements(Collections.emptyList()),
        Collections.emptyList(),
        new Instant(1234L));
    manager.refreshAll();

    TransformWatermarks filteredWatermarks = manager.getWatermarks(graph.getProducer(filtered));
    assertThat(
        filteredWatermarks.getInputWatermark(), not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    assertThat(filteredWatermarks.getOutputWatermark(), not(greaterThan(new Instant(-1000L))));

    CommittedBundle<byte[]> fauxFirstKeyTimerBundle =
        bundleFactory
            .createKeyedBundle(StructuralKey.of("Odd", StringUtf8Coder.of()), impulse)
            .commit(clock.now());
    manager.updateWatermarks(
        fauxFirstKeyTimerBundle,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        fauxFirstKeyTimerBundle.withElements(Collections.emptyList()),
        Collections.emptyList(),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();

    assertThat(filteredWatermarks.getOutputWatermark(), equalTo(new Instant(1234L)));

    CommittedBundle<byte[]> fauxSecondKeyTimerBundle =
        bundleFactory
            .createKeyedBundle(StructuralKey.of("Even", StringUtf8Coder.of()), impulse)
            .commit(clock.now());
    manager.updateWatermarks(
        fauxSecondKeyTimerBundle,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        fauxSecondKeyTimerBundle.withElements(Collections.emptyList()),
        Collections.emptyList(),
        new Instant(5678L));
    manager.refreshAll();
    assertThat(filteredWatermarks.getOutputWatermark(), equalTo(new Instant(5678L)));

    manager.updateWatermarks(
        fauxSecondKeyTimerBundle,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        fauxSecondKeyTimerBundle.withElements(Collections.emptyList()),
        Collections.emptyList(),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    assertThat(
        filteredWatermarks.getOutputWatermark(), not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
  }

  /**
   * Demonstrates that updated output watermarks are monotonic in the presence of late data, when
   * called on an {@link AppliedPTransform} that consumes no input.
   */
  @Test
  public void updateOutputWatermarkShouldBeMonotonic() {
    CommittedBundle<?> firstInput =
        bundleFactory.createBundle(impulse).commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(firstInput),
        new Instant(0L));
    manager.refreshAll();
    TransformWatermarks firstWatermarks = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(firstWatermarks.getOutputWatermark(), equalTo(new Instant(0L)));

    CommittedBundle<?> secondInput =
        bundleFactory.createBundle(impulse).commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(secondInput),
        new Instant(-250L));
    manager.refreshAll();
    TransformWatermarks secondWatermarks = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(secondWatermarks.getOutputWatermark(), not(lessThan(new Instant(0L))));
  }

  /**
   * Demonstrates that updated output watermarks are monotonic in the presence of watermark holds
   * that become earlier than a previous watermark hold.
   */
  @Test
  public void updateWatermarkWithHoldsShouldBeMonotonic() {
    CommittedBundle<byte[]> impulseBundle =
        timestampedBundle(
            impulse,
            TimestampedValue.of(new byte[1], new Instant(1_000_000L)),
            TimestampedValue.of(new byte[2], new Instant(1234L)),
            TimestampedValue.of(new byte[3], new Instant(-1000L)));
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        new Instant(Long.MAX_VALUE));

    CommittedBundle<KV<String, byte[]>> keyBundle =
        timestampedBundle(
            keyed,
            TimestampedValue.of(KV.of("MyKey", new byte[1]), new Instant(1_000_000L)),
            TimestampedValue.of(KV.of("MyKey", new byte[2]), new Instant(1234L)),
            TimestampedValue.of(KV.of("MyKey", new byte[3]), new Instant(-1000L)));
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(keyed),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(keyBundle),
        new Instant(500L));
    manager.refreshAll();
    TransformWatermarks keyedWatermarks = manager.getWatermarks(graph.getProducer(keyed));
    assertThat(
        keyedWatermarks.getInputWatermark(), not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    assertThat(keyedWatermarks.getOutputWatermark(), not(greaterThan(new Instant(500L))));
    Instant oldOutputWatermark = keyedWatermarks.getOutputWatermark();

    TransformWatermarks updatedWatermarks = manager.getWatermarks(graph.getProducer(keyed));
    assertThat(
        updatedWatermarks.getInputWatermark(), not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    // We added a hold prior to the old watermark; we shouldn't progress (due to the earlier hold)
    // but the watermark is monotonic and should not backslide to the new, earlier hold
    assertThat(updatedWatermarks.getOutputWatermark(), equalTo(oldOutputWatermark));
  }

  @Test
  public void updateWatermarkWithUnprocessedElements() {
    WindowedValue<byte[]> first = WindowedValue.valueInGlobalWindow(new byte[1]);
    WindowedValue<byte[]> second =
        WindowedValue.timestampedValueInGlobalWindow(new byte[2], new Instant(-1000L));
    WindowedValue<byte[]> third =
        WindowedValue.timestampedValueInGlobalWindow(new byte[3], new Instant(1234L));
    CommittedBundle<byte[]> impulseBundle =
        bundleFactory.createBundle(impulse).add(first).add(second).add(third).commit(clock.now());

    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);

    CommittedBundle<KV<String, byte[]>> keyBundle =
        timestampedBundle(
            keyed,
            TimestampedValue.of(KV.of("MyKey", new byte[1]), BoundedWindow.TIMESTAMP_MIN_VALUE));
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(keyed),
        impulseBundle.withElements(ImmutableList.of(second, third)),
        Collections.<CommittedBundle<?>>singleton(keyBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    TransformWatermarks keyedWatermarks = manager.getWatermarks(graph.getProducer(keyed));
    // the unprocessed second and third are readded to pending
    assertThat(keyedWatermarks.getInputWatermark(), not(greaterThan(new Instant(-1000L))));
  }

  @Test
  public void updateWatermarkWithCompletedElementsNotPending() {
    WindowedValue<byte[]> first =
        WindowedValue.timestampedValueInGlobalWindow(new byte[1], new Instant(22));
    CommittedBundle<byte[]> impulseBundle =
        bundleFactory.createBundle(impulse).add(first).commit(clock.now());

    WindowedValue<byte[]> second =
        WindowedValue.timestampedValueInGlobalWindow(new byte[2], new Instant(22));
    CommittedBundle<byte[]> neverImpulseBundle =
        bundleFactory.createBundle(impulse).add(second).commit(clock.now());

    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);

    manager.updateWatermarks(
        neverImpulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        neverImpulseBundle.withElements(Collections.emptyList()),
        Collections.emptyList(),
        BoundedWindow.TIMESTAMP_MAX_VALUE);

    manager.refreshAll();
    TransformWatermarks filteredWms = manager.getWatermarks(graph.getProducer(filtered));
    assertThat(filteredWms.getInputWatermark(), equalTo(new Instant(22L)));
  }

  /** Demonstrates that updateWatermarks in the presence of late data is monotonic. */
  @Test
  public void updateWatermarkWithLateData() {
    Instant sourceWatermark = new Instant(1_000_000L);
    CommittedBundle<byte[]> impulseBundle =
        timestampedBundle(
            impulse,
            TimestampedValue.of(new byte[1], sourceWatermark),
            TimestampedValue.of(new byte[2], new Instant(1234L)));

    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        sourceWatermark);

    CommittedBundle<KV<String, byte[]>> keyBundle =
        timestampedBundle(
            keyed,
            TimestampedValue.of(KV.of("MyKey", new byte[1]), sourceWatermark),
            TimestampedValue.of(KV.of("MyKey", new byte[2]), new Instant(1234L)));

    // Finish processing the on-time data. The watermarks should progress to be equal to the source
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(keyed),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(keyBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks onTimeWatermarks = manager.getWatermarks(graph.getProducer(keyed));
    assertThat(onTimeWatermarks.getInputWatermark(), equalTo(sourceWatermark));
    assertThat(onTimeWatermarks.getOutputWatermark(), equalTo(sourceWatermark));

    CommittedBundle<byte[]> lateDataBundle =
        timestampedBundle(impulse, TimestampedValue.of(new byte[3], new Instant(-1000L)));
    // the late data arrives in a downstream PCollection after its watermark has advanced past it;
    // we don't advance the watermark past the current watermark until we've consumed the late data
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(lateDataBundle),
        new Instant(2_000_000L));
    manager.refreshAll();
    TransformWatermarks bufferedLateWm = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(bufferedLateWm.getOutputWatermark(), equalTo(new Instant(2_000_000L)));

    // The input watermark should be held to its previous value (not advanced due to late data; not
    // moved backwards in the presence of watermarks due to monotonicity).
    TransformWatermarks lateDataBufferedWatermark = manager.getWatermarks(graph.getProducer(keyed));
    assertThat(lateDataBufferedWatermark.getInputWatermark(), not(lessThan(sourceWatermark)));
    assertThat(lateDataBufferedWatermark.getOutputWatermark(), not(lessThan(sourceWatermark)));

    CommittedBundle<KV<String, byte[]>> lateKeyedBundle =
        timestampedBundle(
            keyed, TimestampedValue.of(KV.of("MyKey", new byte[3]), new Instant(-1000L)));
    manager.updateWatermarks(
        lateDataBundle,
        TimerUpdate.empty(),
        graph.getProducer(keyed),
        lateDataBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(lateKeyedBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
  }

  @Test
  @Ignore("https://github.com/apache/beam/issues/18826")
  public void updateWatermarkWithDifferentWindowedValueInstances() {
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(
            bundleFactory
                .createBundle(impulse)
                .add(WindowedValue.valueInGlobalWindow(new byte[1]))
                .commit(Instant.now())),
        BoundedWindow.TIMESTAMP_MAX_VALUE);

    CommittedBundle<byte[]> impulseBundle =
        bundleFactory
            .createBundle(impulse)
            .add(WindowedValue.valueInGlobalWindow(new byte[1]))
            .commit(Instant.now());
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(keyed),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.emptyList(),
        null);
    manager.refreshAll();
    TransformWatermarks onTimeWatermarks = manager.getWatermarks(graph.getProducer(keyed));
    assertThat(onTimeWatermarks.getInputWatermark(), equalTo(BoundedWindow.TIMESTAMP_MAX_VALUE));
  }

  /**
   * Demonstrates that after watermarks of an upstream transform are updated, but no output has been
   * produced, the watermarks of a downstream process are advanced.
   */
  @Test
  public void getWatermarksAfterOnlyEmptyOutput() {
    CommittedBundle<byte[]> emptyCreateOutput = multiWindowedBundle(impulse);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(emptyCreateOutput),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks updatedSourceWatermarks = manager.getWatermarks(graph.getProducer(impulse));

    assertThat(
        updatedSourceWatermarks.getOutputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));

    TransformWatermarks finishedFilterWatermarks =
        manager.getWatermarks(graph.getProducer(filtered));
    assertThat(
        finishedFilterWatermarks.getInputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    assertThat(
        finishedFilterWatermarks.getOutputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
  }

  /**
   * Demonstrates that after watermarks of an upstream transform are updated, but no output has been
   * produced, and the downstream transform has a watermark hold, the watermark is held to the hold.
   */
  @Test
  public void getWatermarksAfterHoldAndEmptyOutput() {
    CommittedBundle<byte[]> firstImpulseOutput =
        multiWindowedBundle(impulse, new byte[1], new byte[2]);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(firstImpulseOutput),
        new Instant(12_000L));

    CommittedBundle<byte[]> firstFilterOutput = multiWindowedBundle(filtered);
    manager.updateWatermarks(
        firstImpulseOutput,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        firstImpulseOutput.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(firstFilterOutput),
        new Instant(10_000L));
    manager.refreshAll();
    TransformWatermarks firstFilterWatermarks = manager.getWatermarks(graph.getProducer(filtered));
    assertThat(firstFilterWatermarks.getInputWatermark(), not(lessThan(new Instant(12_000L))));
    assertThat(firstFilterWatermarks.getOutputWatermark(), not(greaterThan(new Instant(10_000L))));

    CommittedBundle<byte[]> emptyImpulseOutput = multiWindowedBundle(impulse);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(emptyImpulseOutput),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks updatedSourceWatermarks = manager.getWatermarks(graph.getProducer(impulse));

    assertThat(
        updatedSourceWatermarks.getOutputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));

    TransformWatermarks finishedFilterWatermarks =
        manager.getWatermarks(graph.getProducer(filtered));
    assertThat(
        finishedFilterWatermarks.getInputWatermark(),
        not(lessThan(BoundedWindow.TIMESTAMP_MAX_VALUE)));
    assertThat(
        finishedFilterWatermarks.getOutputWatermark(), not(greaterThan(new Instant(10_000L))));
  }

  @Test
  public void getSynchronizedProcessingTimeInputWatermarksHeldToPendingBundles() {
    TransformWatermarks watermarks = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(watermarks.getSynchronizedProcessingInputTime(), equalTo(clock.now()));
    assertThat(
        watermarks.getSynchronizedProcessingOutputTime(),
        equalTo(BoundedWindow.TIMESTAMP_MIN_VALUE));

    TransformWatermarks filteredWatermarks = manager.getWatermarks(graph.getProducer(filtered));
    // Non-root processing watermarks don't progress until data has been processed
    assertThat(
        filteredWatermarks.getSynchronizedProcessingInputTime(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));
    assertThat(
        filteredWatermarks.getSynchronizedProcessingOutputTime(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));

    CommittedBundle<byte[]> createOutput =
        bundleFactory.createBundle(impulse).commit(new Instant(1250L));

    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(createOutput),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks createAfterUpdate = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(createAfterUpdate.getSynchronizedProcessingInputTime(), equalTo(clock.now()));
    assertThat(createAfterUpdate.getSynchronizedProcessingOutputTime(), equalTo(clock.now()));

    TransformWatermarks filterAfterProduced = manager.getWatermarks(graph.getProducer(filtered));
    assertThat(
        filterAfterProduced.getSynchronizedProcessingInputTime(), not(greaterThan(clock.now())));
    assertThat(
        filterAfterProduced.getSynchronizedProcessingOutputTime(), not(greaterThan(clock.now())));

    clock.set(new Instant(1500L));
    assertThat(createAfterUpdate.getSynchronizedProcessingInputTime(), equalTo(clock.now()));
    assertThat(createAfterUpdate.getSynchronizedProcessingOutputTime(), equalTo(clock.now()));
    assertThat(
        filterAfterProduced.getSynchronizedProcessingInputTime(),
        not(greaterThan(new Instant(1250L))));
    assertThat(
        filterAfterProduced.getSynchronizedProcessingOutputTime(),
        not(greaterThan(new Instant(1250L))));

    CommittedBundle<?> filterOutputBundle =
        bundleFactory.createBundle(impulseToFlatten).commit(new Instant(1250L));
    manager.updateWatermarks(
        createOutput,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        createOutput.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(filterOutputBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks filterAfterConsumed = manager.getWatermarks(graph.getProducer(filtered));
    assertThat(
        filterAfterConsumed.getSynchronizedProcessingInputTime(),
        not(greaterThan(createAfterUpdate.getSynchronizedProcessingOutputTime())));
    assertThat(
        filterAfterConsumed.getSynchronizedProcessingOutputTime(),
        not(greaterThan(filterAfterConsumed.getSynchronizedProcessingInputTime())));
  }

  /**
   * Demonstrates that the Synchronized Processing Time output watermark cannot progress past
   * pending timers in the same set. This propagates to all downstream SynchronizedProcessingTimes.
   *
   * <p>Also demonstrate that the result is monotonic.
   */
  @Test
  public void getSynchronizedProcessingTimeOutputHeldToPendingTimers() {
    CommittedBundle<byte[]> impulseBundle =
        multiWindowedBundle(impulse, new byte[1], new byte[2], new byte[4], new byte[8]);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        new Instant(1248L));
    manager.refreshAll();

    TransformWatermarks filteredWms = manager.getWatermarks(graph.getProducer(filtered));
    TransformWatermarks filteredDoubledWms =
        manager.getWatermarks(graph.getProducer(filteredNotEmpty));
    Instant initialFilteredWm = filteredWms.getSynchronizedProcessingOutputTime();
    Instant initialFilteredDoubledWm = filteredDoubledWms.getSynchronizedProcessingOutputTime();

    StructuralKey<String> key = StructuralKey.of("key", StringUtf8Coder.of());
    CommittedBundle<byte[]> filteredBundle =
        multiWindowedBundle(filtered, new byte[2], new byte[8]);
    TimerData pastTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(250L),
            new Instant(250L),
            TimeDomain.PROCESSING_TIME);
    TimerData futureTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(4096L),
            new Instant(4096L),
            TimeDomain.PROCESSING_TIME);
    TimerUpdate timers = TimerUpdate.builder(key).setTimer(pastTimer).setTimer(futureTimer).build();
    manager.updateWatermarks(
        impulseBundle,
        timers,
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(filteredBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    Instant startTime = clock.now();
    clock.set(startTime.plus(Duration.millis(250L)));
    // We're held based on the past timer
    assertThat(filteredWms.getSynchronizedProcessingOutputTime(), not(greaterThan(startTime)));
    assertThat(
        filteredDoubledWms.getSynchronizedProcessingOutputTime(), not(greaterThan(startTime)));
    // And we're monotonic
    assertThat(filteredWms.getSynchronizedProcessingOutputTime(), not(lessThan(initialFilteredWm)));
    assertThat(
        filteredDoubledWms.getSynchronizedProcessingOutputTime(),
        not(lessThan(initialFilteredDoubledWm)));

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> firedTimers = manager.extractFiredTimers();
    assertThat(Iterables.getOnlyElement(firedTimers).getTimers(), contains(pastTimer));
    // Our timer has fired, but has not been completed, so it holds our synchronized processing WM
    assertThat(filteredWms.getSynchronizedProcessingOutputTime(), not(greaterThan(startTime)));
    assertThat(
        filteredDoubledWms.getSynchronizedProcessingOutputTime(), not(greaterThan(startTime)));

    CommittedBundle<byte[]> filteredTimerBundle =
        bundleFactory.createKeyedBundle(key, filtered).commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
    CommittedBundle<byte[]> filteredTimerResult =
        bundleFactory
            .createKeyedBundle(key, filteredNotEmpty)
            .commit(filteredWms.getSynchronizedProcessingOutputTime());
    // Complete the processing time timer
    manager.updateWatermarks(
        filteredTimerBundle,
        TimerUpdate.builder(key).withCompletedTimers(Collections.singleton(pastTimer)).build(),
        graph.getProducer(filtered),
        filteredTimerBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(filteredTimerResult),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();

    clock.set(startTime.plus(Duration.millis(500L)));
    assertThat(filteredWms.getSynchronizedProcessingOutputTime(), not(greaterThan(clock.now())));
    // filtered should be held to the time at which the filteredTimerResult fired
    assertThat(
        filteredDoubledWms.getSynchronizedProcessingOutputTime(),
        not(lessThan(filteredTimerResult.getSynchronizedProcessingOutputWatermark())));

    manager.updateWatermarks(
        filteredTimerResult,
        TimerUpdate.empty(),
        graph.getProducer(filteredNotEmpty),
        filteredTimerResult.withElements(Collections.emptyList()),
        Collections.emptyList(),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();

    clock.set(new Instant(Long.MAX_VALUE));

    assertThat(
        filteredDoubledWms.getSynchronizedProcessingOutputTime(),
        not(greaterThan(new Instant(4096))));
  }

  /**
   * Demonstrates that if any earlier processing holds appear in the synchronized processing time
   * output hold the result is monotonic.
   */
  @Test
  public void getSynchronizedProcessingTimeOutputTimeIsMonotonic() {
    Instant startTime = clock.now();
    TransformWatermarks watermarks = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(watermarks.getSynchronizedProcessingInputTime(), equalTo(startTime));

    TransformWatermarks filteredWatermarks = manager.getWatermarks(graph.getProducer(filtered));
    // Non-root processing watermarks don't progress until data has been processed
    assertThat(
        filteredWatermarks.getSynchronizedProcessingInputTime(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));
    assertThat(
        filteredWatermarks.getSynchronizedProcessingOutputTime(),
        not(greaterThan(BoundedWindow.TIMESTAMP_MIN_VALUE)));

    CommittedBundle<byte[]> createOutput =
        bundleFactory.createBundle(impulse).commit(new Instant(1250L));

    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(createOutput),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();
    TransformWatermarks createAfterUpdate = manager.getWatermarks(graph.getProducer(impulse));
    assertThat(
        createAfterUpdate.getSynchronizedProcessingInputTime(), not(greaterThan(clock.now())));
    assertThat(
        createAfterUpdate.getSynchronizedProcessingOutputTime(), not(greaterThan(clock.now())));

    CommittedBundle<byte[]> createSecondOutput =
        bundleFactory.createBundle(impulse).commit(new Instant(750L));
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(createSecondOutput),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();

    assertThat(createAfterUpdate.getSynchronizedProcessingOutputTime(), equalTo(clock.now()));
  }

  @Test
  public void synchronizedProcessingInputTimeIsHeldToUpstreamProcessingTimeTimers() {
    CommittedBundle<byte[]> impulseBundle =
        multiWindowedBundle(impulse, new byte[1], new byte[2], new byte[3]);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        new Instant(40_900L));
    manager.refreshAll();

    CommittedBundle<byte[]> filteredBundle =
        multiWindowedBundle(filtered, new byte[2], new byte[4]);
    Instant upstreamHold = new Instant(2048L);
    TimerData upstreamProcessingTimer =
        TimerData.of(
            StateNamespaces.global(), upstreamHold, upstreamHold, TimeDomain.PROCESSING_TIME);
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.builder(StructuralKey.of("key", StringUtf8Coder.of()))
            .setTimer(upstreamProcessingTimer)
            .build(),
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(filteredBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();

    TransformWatermarks downstreamWms = manager.getWatermarks(graph.getProducer(filteredNotEmpty));
    assertThat(downstreamWms.getSynchronizedProcessingInputTime(), equalTo(clock.now()));

    clock.set(BoundedWindow.TIMESTAMP_MAX_VALUE);
    assertThat(downstreamWms.getSynchronizedProcessingInputTime(), equalTo(upstreamHold));

    manager.extractFiredTimers();
    // Pending processing time timers that have been fired but aren't completed hold the
    // synchronized processing time
    assertThat(downstreamWms.getSynchronizedProcessingInputTime(), equalTo(upstreamHold));

    CommittedBundle<byte[]> otherImpulse =
        multiWindowedBundle(impulse, new byte[4], new byte[8], new byte[12]);
    manager.updateWatermarks(
        otherImpulse,
        TimerUpdate.builder(StructuralKey.of("key", StringUtf8Coder.of()))
            .withCompletedTimers(Collections.singleton(upstreamProcessingTimer))
            .build(),
        graph.getProducer(filtered),
        otherImpulse.withElements(Collections.emptyList()),
        Collections.emptyList(),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();

    assertThat(downstreamWms.getSynchronizedProcessingInputTime(), not(greaterThan(clock.now())));
  }

  @Test
  public void synchronizedProcessingInputTimeIsHeldToPendingBundleTimes() {
    CommittedBundle<byte[]> impulseBundle =
        multiWindowedBundle(impulse, new byte[1], new byte[2], new byte[3]);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.<CommittedBundle<?>>singleton(impulseBundle),
        new Instant(29_919_235L));

    Instant upstreamHold = new Instant(2048L);
    CommittedBundle<byte[]> filteredBundle =
        bundleFactory
            .createKeyedBundle(StructuralKey.of("key", StringUtf8Coder.of()), filtered)
            .commit(upstreamHold);
    manager.updateWatermarks(
        impulseBundle,
        TimerUpdate.empty(),
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(filteredBundle),
        BoundedWindow.TIMESTAMP_MAX_VALUE);
    manager.refreshAll();

    TransformWatermarks downstreamWms = manager.getWatermarks(graph.getProducer(filteredNotEmpty));
    assertThat(downstreamWms.getSynchronizedProcessingInputTime(), equalTo(clock.now()));

    clock.set(BoundedWindow.TIMESTAMP_MAX_VALUE);
    assertThat(downstreamWms.getSynchronizedProcessingInputTime(), equalTo(upstreamHold));
  }

  @Test
  public void extractFiredTimersReturnsFiredEventTimeTimers() {
    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> initialTimers =
        manager.extractFiredTimers();
    // Watermarks haven't advanced
    assertThat(initialTimers, emptyIterable());

    // Advance WM of keyed past the first timer, but ahead of the second and third
    CommittedBundle<byte[]> impulseBundle = multiWindowedBundle(filtered);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.singleton(impulseBundle),
        new Instant(1500L));
    manager.refreshAll();

    TimerData earliestTimer =
        TimerData.of(
            StateNamespaces.global(), new Instant(1000), new Instant(1000), TimeDomain.EVENT_TIME);
    TimerData middleTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(5000L),
            new Instant(5000L),
            TimeDomain.EVENT_TIME);
    TimerData lastTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(10000L),
            new Instant(10000L),
            TimeDomain.EVENT_TIME);
    StructuralKey<byte[]> key = StructuralKey.of(new byte[] {1, 4, 9}, ByteArrayCoder.of());
    TimerUpdate update =
        TimerUpdate.builder(key)
            .setTimer(earliestTimer)
            .setTimer(middleTimer)
            .setTimer(lastTimer)
            .build();

    manager.updateWatermarks(
        impulseBundle,
        update,
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(multiWindowedBundle(impulseToFlatten)),
        new Instant(1000L));
    manager.refreshAll();

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> firstFiredTimers =
        manager.extractFiredTimers();
    assertThat(firstFiredTimers, not(emptyIterable()));
    FiredTimers<AppliedPTransform<?, ?, ?>> firstFired = Iterables.getOnlyElement(firstFiredTimers);
    assertThat(firstFired.getTimers(), contains(earliestTimer));

    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.emptyList(),
        new Instant(50_000L));
    manager.refreshAll();
    assertTrue(manager.extractFiredTimers().isEmpty());

    // confirm processing of the firstExtracted timers
    manager.updateWatermarks(
        null,
        TimerUpdate.builder(key).withCompletedTimers(firstFired.getTimers()).build(),
        graph.getProducer(filtered),
        null,
        Collections.emptyList(),
        new Instant(1000L));

    manager.refreshAll();

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> secondFiredTimers =
        manager.extractFiredTimers();
    assertThat(secondFiredTimers, not(emptyIterable()));
    FiredTimers<AppliedPTransform<?, ?, ?>> secondFired =
        Iterables.getOnlyElement(secondFiredTimers);
    // Contains, in order, middleTimer and then lastTimer
    assertThat(secondFired.getTimers(), contains(middleTimer, lastTimer));
  }

  @Test
  public void extractFiredTimersReturnsFiredProcessingTimeTimers() {
    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> initialTimers =
        manager.extractFiredTimers();
    // Watermarks haven't advanced
    assertThat(initialTimers, emptyIterable());

    // Advance WM of keyed past the first timer, but ahead of the second and third
    CommittedBundle<byte[]> impulseBundle = multiWindowedBundle(filtered);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.singleton(impulseBundle),
        new Instant(1500L));

    TimerData earliestTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(999L),
            new Instant(999L),
            TimeDomain.PROCESSING_TIME);
    TimerData middleTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(5000L),
            new Instant(5000L),
            TimeDomain.PROCESSING_TIME);
    TimerData lastTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(10000L),
            new Instant(10000L),
            TimeDomain.PROCESSING_TIME);
    StructuralKey<?> key = StructuralKey.of(-12L, VarLongCoder.of());
    TimerUpdate update =
        TimerUpdate.builder(key)
            .setTimer(lastTimer)
            .setTimer(earliestTimer)
            .setTimer(middleTimer)
            .build();

    manager.updateWatermarks(
        impulseBundle,
        update,
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(multiWindowedBundle(impulseToFlatten)),
        new Instant(1000L));
    manager.refreshAll();

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> firstFiredTimers =
        manager.extractFiredTimers();
    assertThat(firstFiredTimers, not(emptyIterable()));
    FiredTimers<AppliedPTransform<?, ?, ?>> firstFired = Iterables.getOnlyElement(firstFiredTimers);
    assertThat(firstFired.getTimers(), contains(earliestTimer));

    clock.set(new Instant(50_000L));
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.emptyList(),
        new Instant(50_000L));
    manager.refreshAll();
    assertTrue(manager.extractFiredTimers().isEmpty());

    manager.updateWatermarks(
        null,
        TimerUpdate.builder(key).withCompletedTimers(firstFired.getTimers()).build(),
        graph.getProducer(filtered),
        null,
        Collections.emptyList(),
        new Instant(1000L));

    manager.refreshAll();

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> secondFiredTimers =
        manager.extractFiredTimers();
    assertThat(secondFiredTimers, not(emptyIterable()));
    FiredTimers<AppliedPTransform<?, ?, ?>> secondFired =
        Iterables.getOnlyElement(secondFiredTimers);
    // Contains, in order, middleTimer and then lastTimer
    assertThat(secondFired.getTimers(), contains(middleTimer, lastTimer));
  }

  @Test
  public void extractFiredTimersReturnsFiredSynchronizedProcessingTimeTimers() {
    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> initialTimers =
        manager.extractFiredTimers();
    // Watermarks haven't advanced
    assertThat(initialTimers, emptyIterable());

    // Advance WM of keyed past the first timer, but ahead of the second and third
    CommittedBundle<byte[]> impulseBundle = multiWindowedBundle(filtered);
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.singleton(impulseBundle),
        new Instant(1500L));

    TimerData earliestTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(999L),
            new Instant(999L),
            TimeDomain.SYNCHRONIZED_PROCESSING_TIME);
    TimerData middleTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(5000L),
            new Instant(5000L),
            TimeDomain.SYNCHRONIZED_PROCESSING_TIME);
    TimerData lastTimer =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(10000L),
            new Instant(10000L),
            TimeDomain.SYNCHRONIZED_PROCESSING_TIME);
    StructuralKey<byte[]> key = StructuralKey.of(new byte[] {2, -2, 22}, ByteArrayCoder.of());
    TimerUpdate update =
        TimerUpdate.builder(key)
            .setTimer(lastTimer)
            .setTimer(earliestTimer)
            .setTimer(middleTimer)
            .build();

    manager.updateWatermarks(
        impulseBundle,
        update,
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(multiWindowedBundle(impulseToFlatten)),
        new Instant(1000L));
    manager.refreshAll();

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> firstFiredTimers =
        manager.extractFiredTimers();
    assertThat(firstFiredTimers, not(emptyIterable()));
    FiredTimers<AppliedPTransform<?, ?, ?>> firstFired = Iterables.getOnlyElement(firstFiredTimers);
    assertThat(firstFired.getTimers(), contains(earliestTimer));

    clock.set(new Instant(50_000L));
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.emptyList(),
        new Instant(50_000L));
    manager.refreshAll();
    assertTrue(manager.extractFiredTimers().isEmpty());

    manager.updateWatermarks(
        null,
        TimerUpdate.builder(key).withCompletedTimers(firstFired.getTimers()).build(),
        graph.getProducer(filtered),
        null,
        Collections.emptyList(),
        new Instant(1000L));

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> secondFiredTimers =
        manager.extractFiredTimers();
    assertThat(secondFiredTimers, not(emptyIterable()));
    FiredTimers<AppliedPTransform<?, ?, ?>> secondFired =
        Iterables.getOnlyElement(secondFiredTimers);
    // Contains, in order, middleTimer and then lastTimer
    assertThat(secondFired.getTimers(), contains(middleTimer, lastTimer));
  }

  @Test
  public void processingTimeTimersCanBeReset() {
    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> initialTimers =
        manager.extractFiredTimers();
    assertThat(initialTimers, emptyIterable());

    String timerId = "myTimer";
    StructuralKey<?> key = StructuralKey.of(-12L, VarLongCoder.of());

    TimerData initialTimer =
        TimerData.of(
            timerId,
            StateNamespaces.global(),
            new Instant(5000L),
            new Instant(5000L),
            TimeDomain.PROCESSING_TIME);

    TimerData overridingTimer =
        TimerData.of(
            timerId,
            StateNamespaces.global(),
            new Instant(10000L),
            new Instant(10000L),
            TimeDomain.PROCESSING_TIME);

    TimerUpdate initialUpdate = TimerUpdate.builder(key).setTimer(initialTimer).build();
    TimerUpdate overridingUpdate = TimerUpdate.builder(key).setTimer(overridingTimer).build();

    manager.updateWatermarks(
        null,
        initialUpdate,
        graph.getProducer(impulse),
        null,
        Collections.emptyList(),
        new Instant(5000L));
    manager.refreshAll();

    // This update should override the previous timer.
    manager.updateWatermarks(
        null,
        overridingUpdate,
        graph.getProducer(impulse),
        null,
        Collections.emptyList(),
        new Instant(10000L));

    // Set clock past the timers.
    clock.set(new Instant(50000L));
    manager.refreshAll();

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> firedTimers = manager.extractFiredTimers();
    assertThat(firedTimers, not(emptyIterable()));
    FiredTimers<AppliedPTransform<?, ?, ?>> timers = Iterables.getOnlyElement(firedTimers);
    assertThat(timers.getTimers(), contains(overridingTimer));
  }

  @Test
  public void eventTimeTimersCanBeReset() {
    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> initialTimers =
        manager.extractFiredTimers();
    assertThat(initialTimers, emptyIterable());

    String timerId = "myTimer";
    StructuralKey<?> key = StructuralKey.of(-12L, VarLongCoder.of());

    TimerData initialTimer =
        TimerData.of(
            timerId,
            StateNamespaces.global(),
            new Instant(1000L),
            new Instant(1000L),
            TimeDomain.EVENT_TIME);
    TimerData overridingTimer =
        TimerData.of(
            timerId,
            StateNamespaces.global(),
            new Instant(2000L),
            new Instant(2000L),
            TimeDomain.EVENT_TIME);

    TimerUpdate initialUpdate = TimerUpdate.builder(key).setTimer(initialTimer).build();
    TimerUpdate overridingUpdate = TimerUpdate.builder(key).setTimer(overridingTimer).build();

    CommittedBundle<byte[]> impulseBundle = multiWindowedBundle(filtered);
    manager.updateWatermarks(
        impulseBundle,
        initialUpdate,
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(multiWindowedBundle(impulseToFlatten)),
        new Instant(1000L));
    manager.refreshAll();

    // This update should override the previous timer.
    manager.updateWatermarks(
        impulseBundle,
        overridingUpdate,
        graph.getProducer(filtered),
        impulseBundle.withElements(Collections.emptyList()),
        Collections.<CommittedBundle<?>>singleton(multiWindowedBundle(impulseToFlatten)),
        new Instant(1000L));
    manager.refreshAll();

    // Set WM past the timers.
    manager.updateWatermarks(
        null,
        TimerUpdate.empty(),
        graph.getProducer(impulse),
        null,
        Collections.singleton(impulseBundle),
        new Instant(3000L));
    manager.refreshAll();

    Collection<FiredTimers<AppliedPTransform<?, ?, ?>>> firstFiredTimers =
        manager.extractFiredTimers();
    assertThat(firstFiredTimers, not(emptyIterable()));
    FiredTimers<AppliedPTransform<?, ?, ?>> firstFired = Iterables.getOnlyElement(firstFiredTimers);
    assertThat(firstFired.getTimers(), contains(overridingTimer));
  }

  @Test
  public void inputWatermarkDuplicates() {
    Watermark mockWatermark = Mockito.mock(Watermark.class);

    AppliedPTransformInputWatermark underTest =
        new AppliedPTransformInputWatermark(
            "underTest", ImmutableList.of(mockWatermark), update -> {});

    // Refresh
    when(mockWatermark.get()).thenReturn(new Instant(0));
    underTest.refresh();
    assertEquals(new Instant(0), underTest.get());

    // Apply a timer update
    StructuralKey<String> key = StructuralKey.of("key", StringUtf8Coder.of());
    TimerData timer1 =
        TimerData.of(
            "a",
            StateNamespaces.global(),
            new Instant(100),
            new Instant(100),
            TimeDomain.EVENT_TIME);
    TimerData timer2 =
        TimerData.of(
            "a",
            StateNamespaces.global(),
            new Instant(200),
            new Instant(200),
            TimeDomain.EVENT_TIME);
    underTest.updateTimers(TimerUpdate.builder(key).setTimer(timer1).setTimer(timer2).build());

    // Only the last timer update should be observable
    assertEquals(timer2.getTimestamp(), underTest.getEarliestTimerTimestamp());

    // Advance the input watermark
    when(mockWatermark.get()).thenReturn(new Instant(1000));
    underTest.refresh();
    assertEquals(new Instant(1000), underTest.get()); // input watermark is not held by timers

    // Examine the fired event time timers
    Map<StructuralKey<?>, List<TimerData>> fired = underTest.extractFiredEventTimeTimers();
    List<TimerData> timers = fired.get(key);
    assertNotNull(timers);
    assertThat(timers, contains(timer2));

    // Update based on timer firings
    underTest.updateTimers(TimerUpdate.builder(key).withCompletedTimers(timers).build());

    // Now we should be able to advance
    assertEquals(BoundedWindow.TIMESTAMP_MAX_VALUE, underTest.getEarliestTimerTimestamp());

    // Nothing left to fire
    fired = underTest.extractFiredEventTimeTimers();
    assertThat(fired.entrySet(), empty());
  }

  @Test
  public void timerUpdateBuilderBuildAddsAllAddedTimers() {
    TimerData set =
        TimerData.of(
            StateNamespaces.global(), new Instant(10L), new Instant(10L), TimeDomain.EVENT_TIME);
    TimerData deleted =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(24L),
            new Instant(24L),
            TimeDomain.PROCESSING_TIME);
    TimerData completedOne =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(1024L),
            new Instant(1024L),
            TimeDomain.SYNCHRONIZED_PROCESSING_TIME);
    TimerData completedTwo =
        TimerData.of(
            StateNamespaces.global(),
            new Instant(2048L),
            new Instant(2048L),
            TimeDomain.EVENT_TIME);

    TimerUpdate update =
        TimerUpdate.builder(StructuralKey.of("foo", StringUtf8Coder.of()))
            .withCompletedTimers(ImmutableList.of(completedOne, completedTwo))
            .setTimer(set)
            .deletedTimer(deleted)
            .build();

    assertThat(update.getCompletedTimers(), containsInAnyOrder(completedOne, completedTwo));
    assertThat(update.getSetTimers(), contains(set));
    assertThat(update.getDeletedTimers(), contains(deleted));
  }

  @Test
  public void timerUpdateBuilderWithSetAtEndOfTime() {
    Instant timerStamp = BoundedWindow.TIMESTAMP_MAX_VALUE;
    TimerData tooFar =
        TimerData.of(StateNamespaces.global(), timerStamp, timerStamp, TimeDomain.EVENT_TIME);

    TimerUpdateBuilder builder = TimerUpdate.builder(StructuralKey.empty());
    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage(timerStamp.toString());
    builder.setTimer(tooFar);
  }

  @Test
  public void timerUpdateBuilderWithSetPastEndOfTime() {
    Instant timerStamp = BoundedWindow.TIMESTAMP_MAX_VALUE.plus(Duration.standardMinutes(2));
    TimerData tooFar =
        TimerData.of(StateNamespaces.global(), timerStamp, timerStamp, TimeDomain.EVENT_TIME);

    TimerUpdateBuilder builder = TimerUpdate.builder(StructuralKey.empty());
    thrown.expect(IllegalArgumentException.class);
    thrown.expectMessage(timerStamp.toString());
    builder.setTimer(tooFar);
  }

  @Test
  public void timerUpdateBuilderWithSetThenDeleteHasOnlyDeleted() {
    TimerUpdateBuilder builder = TimerUpdate.builder(null);
    Instant now = Instant.now();
    TimerData timer = TimerData.of(StateNamespaces.global(), now, now, TimeDomain.EVENT_TIME);

    TimerUpdate built = builder.setTimer(timer).deletedTimer(timer).build();

    assertThat(built.getSetTimers(), emptyIterable());
    assertThat(built.getDeletedTimers(), contains(timer));
  }

  @Test
  public void timerUpdateBuilderWithDeleteThenSetHasOnlySet() {
    TimerUpdateBuilder builder = TimerUpdate.builder(null);
    Instant now = Instant.now();
    TimerData timer = TimerData.of(StateNamespaces.global(), now, now, TimeDomain.EVENT_TIME);

    TimerUpdate built = builder.deletedTimer(timer).setTimer(timer).build();

    assertThat(built.getSetTimers(), contains(timer));
    assertThat(built.getDeletedTimers(), emptyIterable());
  }

  @Test
  public void timerUpdateBuilderWithSetAfterBuildNotAddedToBuilt() {
    TimerUpdateBuilder builder = TimerUpdate.builder(null);
    Instant now = Instant.now();
    TimerData timer = TimerData.of(StateNamespaces.global(), now, now, TimeDomain.EVENT_TIME);

    TimerUpdate built = builder.build();
    builder.setTimer(timer);
    assertThat(built.getSetTimers(), emptyIterable());
    builder.build();
    assertThat(built.getSetTimers(), emptyIterable());
  }

  @Test
  public void timerUpdateBuilderWithDeleteAfterBuildNotAddedToBuilt() {
    TimerUpdateBuilder builder = TimerUpdate.builder(null);
    Instant now = Instant.now();
    TimerData timer = TimerData.of(StateNamespaces.global(), now, now, TimeDomain.EVENT_TIME);

    TimerUpdate built = builder.build();
    builder.deletedTimer(timer);
    assertThat(built.getDeletedTimers(), emptyIterable());
    builder.build();
    assertThat(built.getDeletedTimers(), emptyIterable());
  }

  @Test
  public void timerUpdateBuilderWithCompletedAfterBuildNotAddedToBuilt() {
    TimerUpdateBuilder builder = TimerUpdate.builder(null);
    Instant now = Instant.now();
    TimerData timer = TimerData.of(StateNamespaces.global(), now, now, TimeDomain.EVENT_TIME);

    TimerUpdate built = builder.build();
    builder.withCompletedTimers(ImmutableList.of(timer));
    assertThat(built.getCompletedTimers(), emptyIterable());
    builder.build();
    assertThat(built.getCompletedTimers(), emptyIterable());
  }

  @Test
  public void timerUpdateWithCompletedTimersNotAddedToExisting() {
    TimerUpdateBuilder builder = TimerUpdate.builder(null);
    Instant now = Instant.now();
    TimerData timer = TimerData.of(StateNamespaces.global(), now, now, TimeDomain.EVENT_TIME);

    TimerUpdate built = builder.build();
    assertThat(built.getCompletedTimers(), emptyIterable());
    assertThat(
        built.withCompletedTimers(ImmutableList.of(timer)).getCompletedTimers(), contains(timer));
    assertThat(built.getCompletedTimers(), emptyIterable());
  }

  @SafeVarargs
  private final <T> CommittedBundle<T> timestampedBundle(
      PCollection<T> pc, TimestampedValue<T>... values) {
    UncommittedBundle<T> bundle = bundleFactory.createBundle(pc);
    for (TimestampedValue<T> value : values) {
      bundle.add(
          WindowedValue.timestampedValueInGlobalWindow(value.getValue(), value.getTimestamp()));
    }
    return bundle.commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
  }

  @SafeVarargs
  private final <T> CommittedBundle<T> multiWindowedBundle(PCollection<T> pc, T... values) {
    UncommittedBundle<T> bundle = bundleFactory.createBundle(pc);
    Collection<BoundedWindow> windows =
        ImmutableList.of(
            GlobalWindow.INSTANCE,
            new IntervalWindow(BoundedWindow.TIMESTAMP_MIN_VALUE, new Instant(0)));
    for (T value : values) {
      bundle.add(
          WindowedValue.of(value, BoundedWindow.TIMESTAMP_MIN_VALUE, windows, PaneInfo.NO_FIRING));
    }
    return bundle.commit(BoundedWindow.TIMESTAMP_MAX_VALUE);
  }
}
