@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 shouldReturnValuesForOpenEndBounds_fetchAll() {
    // Given:
    final Range<Instant> end = Range.open(
      NOW,
      NOW.plusSeconds(10)
    );

    final Range<Instant> startEquiv = Range.open(
      end.lowerEndpoint().minus(WINDOW_SIZE),
      end.upperEndpoint().minus(WINDOW_SIZE)
    );

    final StateQueryResult<KeyValueIterator<Windowed<GenericKey>, ValueAndTimestamp<GenericRow>>> partitionResult = new StateQueryResult<>();
    final QueryResult<KeyValueIterator<Windowed<GenericKey>, ValueAndTimestamp<GenericRow>>> queryResult = QueryResult.forResult(keyValueIterator);
    queryResult.setPosition(POSITION);
    partitionResult.addResult(PARTITION, queryResult);
    when(kafkaStreams.query(any(StateQueryRequest.class))).thenReturn(partitionResult);

    when(keyValueIterator.hasNext())
      .thenReturn(true, true, true, false);

    when(keyValueIterator.next())
      .thenReturn(new KeyValue<>(new Windowed<>(A_KEY,
        new TimeWindow(startEquiv.lowerEndpoint().toEpochMilli(),
          startEquiv.lowerEndpoint().toEpochMilli() + WINDOW_SIZE.toMillis())), VALUE_1))
      .thenReturn(new KeyValue<>(new Windowed<>(A_KEY2,
        new TimeWindow(startEquiv.lowerEndpoint().plusMillis(1).toEpochMilli(),
          startEquiv.lowerEndpoint().toEpochMilli() + WINDOW_SIZE.toMillis() + 1)), VALUE_2))
      .thenReturn(new KeyValue<>(new Windowed<>(A_KEY3,
        new TimeWindow(startEquiv.upperEndpoint().toEpochMilli(),
          startEquiv.upperEndpoint().toEpochMilli() + WINDOW_SIZE.toMillis())), VALUE_3))
      .thenThrow(new AssertionError());


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

    // Then:
    final Iterator<WindowedRow> rowIterator = result.getRowIterator();
    assertThat(rowIterator.hasNext(), is(true));
    assertThat(rowIterator.next(),
      is (WindowedRow.of(
        SCHEMA,
        windowedKey(A_KEY2, startEquiv.lowerEndpoint().plusMillis(1)),
        VALUE_2.value(),
        VALUE_2.timestamp())));
    assertThat(rowIterator.hasNext(), is(false));
    assertThat(result.getPosition(), not(Optional.empty()));
    assertThat(result.getPosition().get(), is(POSITION));
  }