### Question:
SessionWindowExpression 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(SessionWindows.with(sizeUnit.toMillis(gap))) .aggregate(initializer, aggregator, aggregator.getMerger(), materialized); } SessionWindowExpression(long gap, TimeUnit sizeUnit); private  SessionWindowExpression(Optional<NodeLocation> location, long gap,
                                  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:
@SuppressWarnings("unchecked") @Test public void shouldCreateSessionWindowedStreamWithInactiviyGap() { EasyMock.expect(groupedStreamMock.windowedBy(EasyMock.capture(sessionWindows))).andReturn(sessionWindowed); EasyMock.expect(sessionWindowed.aggregate(same(initializer), same(aggregator), anyObject(Merger.class), same(materialized))).andReturn(null); EasyMock.replay(groupedStreamMock, aggregator, sessionWindowed); expression.applyAggregate(groupedStreamMock, initializer, aggregator, materialized); assertThat(sessionWindows.getValue().inactivityGap(), equalTo(5000L)); EasyMock.verify(groupedStreamMock); }
@SuppressWarnings("unchecked") @Test public void shouldGetMergerForSessionWindowsFromUdafAggregator() { EasyMock.expect(groupedStreamMock.windowedBy(EasyMock.capture(sessionWindows))).andReturn(sessionWindowed); EasyMock.expect(sessionWindowed.aggregate(same(initializer), same(aggregator), same(merger), same(materialized))).andReturn(null); EasyMock.expect(aggregator.getMerger()).andReturn(merger); EasyMock.replay(groupedStreamMock, aggregator, sessionWindowed); expression.applyAggregate(groupedStreamMock, initializer, aggregator, materialized); EasyMock.verify(aggregator); }