@SuppressWarnings("OptionalGetWithoutIsPresent") // Enforced by type
  @Override
  public StreamsMaterializedWindowedTable windowed() {
    if (!windowInfo.isPresent()) {
      throw new UnsupportedOperationException("Table has non-windowed key");
    }

    final WindowInfo wndInfo = windowInfo.get();
    final WindowType wndType = wndInfo.getType();
    switch (wndType) {
      case SESSION:
        return new KsMaterializedSessionTable(stateStore,
            SessionStoreCacheBypass::fetch, SessionStoreCacheBypass::fetchRange);

      case HOPPING:
      case TUMBLING:
        return new KsMaterializedWindowTable(stateStore, wndInfo.getSize().get(),
            WindowStoreCacheBypass::fetch,
            WindowStoreCacheBypass::fetchAll,
            WindowStoreCacheBypass::fetchRange);

      default:
        throw new UnsupportedOperationException("Unknown window type: " + wndInfo);
    }
  }
@Test
  public void shouldReturnWindowedForHopping() {
    // Given:
    givenWindowType(Optional.of(WindowType.HOPPING));

    // When:
    final StreamsMaterializedWindowedTable table = materialization.windowed();

    // Then:
    assertThat(table, is(instanceOf(KsMaterializedWindowTable.class)));
  }