@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 shouldGetWithCorrectParams() {
    // When:
    table.get(A_KEY, PARTITION);

    // Then:
    verify(tableStore).get(A_KEY);
  }