### Question:
TumblingWindowExpression extends KsqlWindowExpression { @SuppressWarnings("unchecked") @Override public KTable applyAggregate(final KGroupedStream groupedStream, final Initializer initializer, final UdafAggregator aggregator, final Materialized<String, GenericRow, ?> materialized) { return groupedStream.windowedBy(TimeWindows.of(sizeUnit.toMillis(size))) .aggregate(initializer, aggregator, materialized); } TumblingWindowExpression(long size, TimeUnit sizeUnit); private  TumblingWindowExpression(Optional<NodeLocation> location, long size,
                                   TimeUnit sizeUnit); @Override String toString(); @Override int hashCode(); @Override boolean equals(Object o); @SuppressWarnings("unchecked") @Override KTable applyAggregate(final KGroupedStream groupedStream,
                               final Initializer initializer,
                               final UdafAggregator aggregator,
                               final Materialized<String, GenericRow, ?> materialized);  }### Answer:
@Test public void shouldCreateTumblingWindowAggregate() { final KGroupedStream stream = EasyMock.createNiceMock(KGroupedStream.class); final TimeWindowedKStream windowedKStream = EasyMock.createNiceMock(TimeWindowedKStream.class); final UdafAggregator aggregator = EasyMock.createNiceMock(UdafAggregator.class); final TumblingWindowExpression windowExpression = new TumblingWindowExpression(10, TimeUnit.SECONDS); final Initializer initializer = () -> 0; final Materialized<String, GenericRow, WindowStore<Bytes, byte[]>> store = Materialized.as("store"); EasyMock.expect(stream.windowedBy(TimeWindows.of(10000L))).andReturn(windowedKStream); EasyMock.expect(windowedKStream.aggregate(same(initializer), same(aggregator), same(store))).andReturn(null); EasyMock.replay(stream, windowedKStream); windowExpression.applyAggregate(stream, initializer, aggregator, store); EasyMock.verify(stream, windowedKStream); }