@Override
  public KsMaterializedQueryResult<WindowedRow> get(
      final GenericKey key,
      final int partition,
      final Range<Instant> windowStart,
      final Range<Instant> windowEnd,
      final Optional<Position> position
  ) {
    try {
      final ReadOnlySessionStore<GenericKey, GenericRow> store = stateStore
          .store(QueryableStoreTypes.sessionStore(), partition);

      return KsMaterializedQueryResult.rowIterator(
          findSession(store, key, windowStart, windowEnd).iterator());
    } catch (final Exception e) {
      throw new MaterializationException("Failed to get value from materialized table", e);
    }
  }
@Test
  public void shouldThrowIfStoreFetchFails() {
    // Given:
    when(cacheBypassFetcher.fetch(any(), any()))
        .thenThrow(new MaterializationTimeOutException("Boom"));

    // When:
    final Exception e = assertThrows(
        MaterializationException.class,
        () -> table.get(A_KEY, PARTITION, WINDOW_START_BOUNDS, WINDOW_END_BOUNDS)
    );

    // Then:
    assertThat(e.getMessage(), containsString(
        "Failed to get value from materialized table"));
    assertThat(e.getCause(), (instanceOf(MaterializationTimeOutException.class)));
  }