@Override
  public KsMaterializedQueryResult<Row> get(
      final GenericKey key,
      final int partition,
      final Optional<Position> position
  ) {
    try {
      final ReadOnlyKeyValueStore<GenericKey, ValueAndTimestamp<GenericRow>> store = stateStore
          .store(QueryableStoreTypes.timestampedKeyValueStore(), partition);

      final ValueAndTimestamp<GenericRow> row = store.get(key);
      if (row == null) {
        return KsMaterializedQueryResult.rowIterator(Collections.emptyIterator());
      } else {
        return KsMaterializedQueryResult.rowIterator(ImmutableList.of(Row.of(
            stateStore.schema(), key, row.value(), row.timestamp())).iterator());
      }
    } catch (final Exception e) {
      throw new MaterializationException("Failed to get value from materialized table", e);
    }
  }
@Test
  public void shouldThrowIfStoreGetFails() {
    // Given:
    when(tableStore.get(any())).thenThrow(new MaterializationTimeOutException("Boom"));

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

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