@Nullable
  @Override
  public <InputT> TransformEvaluator<InputT> forApplication(
      AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle) {
    return (TransformEvaluator<InputT>) new ImpulseEvaluator(ctxt, (AppliedPTransform) application);
  }
@Test
  public void testImpulse() throws Exception {
    Pipeline p = Pipeline.create();
    PCollection<byte[]> impulseOut = p.apply(Impulse.create());

    AppliedPTransform<?, ?, ?> impulseApplication = DirectGraphs.getProducer(impulseOut);

    ImpulseEvaluatorFactory factory = new ImpulseEvaluatorFactory(context);

    WindowedValue<ImpulseShard> inputShard = WindowedValue.valueInGlobalWindow(new ImpulseShard());
    CommittedBundle<ImpulseShard> inputShardBundle =
        bundleFactory.<ImpulseShard>createRootBundle().add(inputShard).commit(Instant.now());

    when(context.createBundle(impulseOut)).thenReturn(bundleFactory.createBundle(impulseOut));
    TransformEvaluator<ImpulseShard> evaluator =
        factory.forApplication(impulseApplication, inputShardBundle);
    evaluator.processElement(inputShard);
    TransformResult<ImpulseShard> result = evaluator.finishBundle();
    assertThat(
        "Exactly one output from a single ImpulseShard",
        Iterables.size(result.getOutputBundles()),
        equalTo(1));
    UncommittedBundle<?> outputBundle = result.getOutputBundles().iterator().next();
    CommittedBundle<?> committedOutputBundle = outputBundle.commit(Instant.now());
    assertThat(
        committedOutputBundle.getMinimumTimestamp(), equalTo(BoundedWindow.TIMESTAMP_MIN_VALUE));
    assertThat(committedOutputBundle.getPCollection(), equalTo(impulseOut));
    assertThat(
        "Should only be one impulse element",
        Iterables.size(committedOutputBundle.getElements()),
        equalTo(1));
    assertThat(
        committedOutputBundle.getElements().iterator().next().getWindows(),
        contains(GlobalWindow.INSTANCE));
    assertArrayEquals(
        "Output should be an empty byte array",
        new byte[0],
        (byte[]) committedOutputBundle.getElements().iterator().next().getValue());
  }