@Override
  public KsMaterializedQueryResult<Row> get(
      final GenericKey key,
      final int partition,
      final Optional<Position> position
  ) {
    try {
      final KeyQuery<GenericKey, ValueAndTimestamp<GenericRow>> query = KeyQuery.withKey(key);
      StateQueryRequest<ValueAndTimestamp<GenericRow>>
          request = inStore(stateStore.getStateStoreName())
          .withQuery(query)
          .withPartitions(ImmutableSet.of(partition));
      if (position.isPresent()) {
        request = request.withPositionBound(PositionBound.at(position.get()));
      }
      final StateQueryResult<ValueAndTimestamp<GenericRow>>
          result = stateStore.getKafkaStreams().query(request);
      final QueryResult<ValueAndTimestamp<GenericRow>> queryResult =
          result.getPartitionResults().get(partition);
      // Some of these failures are retriable, and in the future, we may want to retry
      // locally before throwing.
      if (queryResult.isFailure()) {
        throw failedQueryException(queryResult);
      } else if (queryResult.getResult() == null) {
        return KsMaterializedQueryResult.rowIteratorWithPosition(
            Collections.emptyIterator(), queryResult.getPosition());
      } else {
        final ValueAndTimestamp<GenericRow> row = queryResult.getResult();
        return KsMaterializedQueryResult.rowIteratorWithPosition(
            ImmutableList.of(Row.of(stateStore.schema(), key, row.value(), row.timestamp()))
                .iterator(),
            queryResult.getPosition());
      }
    } catch (final NotUpToBoundException | MaterializationException e) {
      throw e;
    } catch (final Exception e) {
      throw new MaterializationException("Failed to get value from materialized table", e);
    }
  }
@Test
  public void shouldRangeQueryWithCorrectParams_noBounds() {
    // Given:
    when(kafkaStreams.query(any())).thenReturn(getIteratorResult());

    // When:
    table.get(PARTITION, null, null);

    // Then:
    verify(kafkaStreams).query(queryTypeCaptor.capture());
    StateQueryRequest request = queryTypeCaptor.getValue();
    assertThat(request.getQuery(), instanceOf(RangeQuery.class));
    RangeQuery rangeQuery = (RangeQuery)request.getQuery();
    assertThat(rangeQuery.getLowerBound(), is(Optional.empty()));
    assertThat(rangeQuery.getUpperBound(), is(Optional.empty()));
  }