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