@Override
  public KsMaterializedQueryResult<WindowedRow> get(
      final GenericKey key,
      final int partition,
      final Range<Instant> windowStartBounds,
      final Range<Instant> windowEndBounds,
      final Optional<Position> position
  ) {
    try {
      final Instant lower = calculateLowerBound(windowStartBounds, windowEndBounds);
      final Instant upper = calculateUpperBound(windowStartBounds, windowEndBounds);
      final WindowKeyQuery<GenericKey, ValueAndTimestamp<GenericRow>> query =
          WindowKeyQuery.withKeyAndWindowStartRange(key, lower, upper);
      StateQueryRequest<WindowStoreIterator<ValueAndTimestamp<GenericRow>>> request =
          inStore(stateStore.getStateStoreName()).withQuery(query);
      if (position.isPresent()) {
        request = request.withPositionBound(PositionBound.at(position.get()));
      }
      final KafkaStreams streams = stateStore.getKafkaStreams();
      final StateQueryResult<WindowStoreIterator<ValueAndTimestamp<GenericRow>>> result =
          streams.query(request);

      final QueryResult<WindowStoreIterator<ValueAndTimestamp<GenericRow>>> queryResult =
          result.getPartitionResults().get(partition);

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

      if (queryResult.getResult() == null) {
        return KsMaterializedQueryResult.rowIteratorWithPosition(
            Collections.emptyIterator(), queryResult.getPosition());
      }

      try (WindowStoreIterator<ValueAndTimestamp<GenericRow>> it
          = queryResult.getResult()) {

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

        while (it.hasNext()) {
          final KeyValue<Long, ValueAndTimestamp<GenericRow>> next = it.next();

          final Instant windowStart = Instant.ofEpochMilli(next.key);
          if (!windowStartBounds.contains(windowStart)) {
            continue;
          }

          final Instant windowEnd = windowStart.plus(windowSize);
          if (!windowEndBounds.contains(windowEnd)) {
            continue;
          }

          final TimeWindow window =
              new TimeWindow(windowStart.toEpochMilli(), windowEnd.toEpochMilli());

          final WindowedRow row = WindowedRow.of(
              stateStore.schema(),
              new Windowed<>(key, window),
              next.value.value(),
              next.value.timestamp()
          );

          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
  @SuppressWarnings("unchecked")
  public void shouldReturnValuesForClosedStartBounds() {
    // Given:
    final Range<Instant> start = Range.closed(
      Instant.ofEpochMilli(System.currentTimeMillis()),
      NOW.plusSeconds(10)
    );

    final StateQueryResult<WindowStoreIterator<ValueAndTimestamp<GenericRow>>> partitionResult = new StateQueryResult<>();
    final QueryResult<WindowStoreIterator<ValueAndTimestamp<GenericRow>>> result = QueryResult.forResult(fetchIterator);
    result.setPosition(POSITION);
    partitionResult.addResult(PARTITION, result);
    when(kafkaStreams.query(any(StateQueryRequest.class))).thenReturn(partitionResult);
    when(fetchIterator.hasNext()).thenReturn(true, true, false);

    when(fetchIterator.next())
      .thenReturn(new KeyValue<>(start.lowerEndpoint().toEpochMilli(), VALUE_1))
      .thenReturn(new KeyValue<>(start.upperEndpoint().toEpochMilli(), VALUE_2))
      .thenThrow(new AssertionError());

    // When:
    final Iterator<WindowedRow> rowIterator =
        table.get(A_KEY, PARTITION, start, Range.all()).rowIterator;

    // Then:
    assertThat(rowIterator.hasNext(), is(true));
    final List<WindowedRow> resultList = Lists.newArrayList(rowIterator);
    assertThat(resultList, contains(
      WindowedRow.of(
        SCHEMA,
        windowedKey(start.lowerEndpoint()),
        VALUE_1.value(),
        VALUE_1.timestamp()
      ),
      WindowedRow.of(
        SCHEMA,
        windowedKey(start.upperEndpoint()),
        VALUE_2.value(),
        VALUE_2.timestamp()
      )
    ));
  }