### Question:
TimestampedKeyValueGetQuery extends KeyedLocalStoreQuery<K, K, V> { @Override public Try<List<KV<K, V>>> execute(final KafkaStreamsContainer container, final long limit) { LocalStoreAccessor<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>> accessor = container.localTimestampedKeyValueStore(storeName()); final Reader<ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>, List<KV<K, V>>> reader = reader(key()).map(value -> Optional.ofNullable(value) .map(v -> Collections.singletonList(KV.of(key(), v.value(), v.timestamp()))) .orElse(Collections.emptyList())); return new LocalStoreQueryExecutor<>(accessor).execute(reader); } TimestampedKeyValueGetQuery(final String storeName,
                                final K key,
                                final Serializer<K> keySerializer); @Override StoreType storeType(); @Override StoreOperation operationType(); @Override Try<List<KV<K, V>>> execute(final KafkaStreamsContainer container, final long limit);  }### Answer:
@Test public void shouldGetGivenKey() { TimestampedKeyValueGetQuery<String, String> query = new TimestampedKeyValueGetQuery<>(STORE_NAME, "key", Serdes.String().serializer()); final var mkContainer = Mockito.mock(KafkaStreamsContainer.class); ReadOnlyKeyValueStore store = mock(ReadOnlyKeyValueStore.class); when(store.get("key")).thenReturn(ValueAndTimestamp.make("value", 1L)); when(mkContainer.localTimestampedKeyValueStore(matches(STORE_NAME))).thenReturn(new LocalStoreAccessor<>(() -> store)); Try<List<KV<String, String>>> result = query.execute(mkContainer); Assertions.assertEquals(1, result.get().size()); Assertions.assertEquals(KV.of("key", "value", 1L), result.get().get(0)); }