@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 WindowRangeQuery<GenericKey, GenericRow> query = WindowRangeQuery.withKey(key);
      StateQueryRequest<KeyValueIterator<Windowed<GenericKey>, GenericRow>> request =
          inStore(stateStore.getStateStoreName()).withQuery(query);
      if (position.isPresent()) {
        request = request.withPositionBound(PositionBound.at(position.get()));
      }
      final StateQueryResult<KeyValueIterator<Windowed<GenericKey>, GenericRow>> result =
          stateStore.getKafkaStreams().query(request);

      final QueryResult<KeyValueIterator<Windowed<GenericKey>, GenericRow>> queryResult =
          result.getPartitionResults().get(partition);

      if (queryResult.isFailure()) {
        throw failedQueryException(queryResult);
      }

      try (KeyValueIterator<Windowed<GenericKey>, GenericRow> it =
          queryResult.getResult()) {

        final Builder<WindowedRow> builder = ImmutableList.builder();

        while (it.hasNext()) {
          final KeyValue<Windowed<GenericKey>, GenericRow> next = it.next();
          final Window wnd = next.key.window();

          if (!windowStart.contains(wnd.startTime())) {
            continue;
          }

          if (!windowEnd.contains(wnd.endTime())) {
            continue;
          }

          final long rowTime = wnd.end();

          final WindowedRow row = WindowedRow.of(
              stateStore.schema(),
              next.key,
              next.value,
              rowTime
          );

          builder.add(row);
        }
        return KsMaterializedQueryResult.rowIteratorWithPosition(
            builder.build().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 shouldReturnValueIfSessionStartsAtUpperBoundIfUpperBoundClosed() {
    // Given:
    final Range<Instant> startBounds = Range.closed(
      LOWER_INSTANT,
      UPPER_INSTANT
    );

    final Instant wend = UPPER_INSTANT.plusMillis(1);
    givenSingleSession(UPPER_INSTANT, wend);

    // When:
    final KsMaterializedQueryResult<WindowedRow> result =
        table.get(A_KEY, PARTITION, startBounds, Range.all());

    // Then:
    final Iterator<WindowedRow> rowIterator = result.getRowIterator();
    assertThat(rowIterator.hasNext(), is(true));
    assertThat(rowIterator.next(), is(
        WindowedRow.of(
            SCHEMA,
            sessionKey(UPPER_INSTANT, wend),
            A_VALUE,
            wend.toEpochMilli()
        )
    ));
    assertThat(result.getPosition(), not(Optional.empty()));
    assertThat(result.getPosition().get(), is(POSITION));
  }