public static KTableHolder<GenericKey> build(
      final KGroupedStreamHolder groupedStream,
      final StreamAggregate aggregate,
      final RuntimeBuildContext buildContext,
      final MaterializedFactory materializedFactory) {
    return build(
        groupedStream,
        aggregate,
        buildContext,
        materializedFactory,
        new AggregateParamsFactory()
    );
  }
@Test
  @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT")
  public void shouldBuildTumblingWindowedAggregateCorrectly() {
    // Given:
    givenTumblingWindowedAggregate();

    // When:
    final KTableHolder<Windowed<GenericKey>> result = windowedAggregate.build(planBuilder, planInfo);

    // Then:
    assertThat(result.getTable(), is(windowedWithWindowBounds));
    verify(gracePeriodClause).toDuration();
    verify(retentionClause).toDuration();

    final InOrder inOrder = Mockito.inOrder(
        groupedStream,
        timeWindowedStream,
        windowed,
        windowedWithResults,
        windowedWithWindowBounds
    );
    inOrder.verify(groupedStream).windowedBy(TimeWindows.ofSizeAndGrace(WINDOW, gracePeriodClause.toDuration()));
    inOrder.verify(timeWindowedStream).aggregate(initializer, aggregator, timeWindowMaterialized);
    inOrder.verify(windowed).transformValues(any(), any(Named.class));
    inOrder.verify(windowedWithResults).transformValues(any(), any(Named.class));
    inOrder.verifyNoMoreInteractions();

    assertThat(result.getTable(), is(windowedWithWindowBounds));
  }