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
  public void shouldBuildAggregatorParamsCorrectlyForWindowedAggregate() {
    for (final Runnable given : given()) {
      // Given:
      clearInvocations(
          groupedStream,
          timeWindowedStream,
          sessionWindowedStream,
          aggregated,
          aggregateParamsFactory
      );
      when(aggregateParamsFactory.create(any(), any(), any(), any(), anyBoolean(), any()))
          .thenReturn(aggregateParams);
      given.run();

      // When:
      windowedAggregate.build(planBuilder, planInfo);

      // Then:
      verify(aggregateParamsFactory)
          .create(INPUT_SCHEMA, NON_AGG_COLUMNS, functionRegistry, FUNCTIONS, true, KsqlConfig.empty());
    }
  }