Skip to content

KAFKA-9243 Update the javadocs to include TimestampKeyValueStore #7848

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
miroswan wants to merge 2 commits into from
Closed

KAFKA-9243 Update the javadocs to include TimestampKeyValueStore #7848

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
337c88f
KAFKA-9243 Update the javadocs for KTable.java
miroswan Dec 18, 2019
508ed7d
Updating javadocs code snippets for KTables to use types that compile
miroswan Dec 18, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 11 additions & 11 deletions streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.QueryableStoreType;
import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.state.TimestampedKeyValueStore;

import java.util.function.Function;

Expand All @@ -43,7 +43,7 @@
* A {@code KTable} can be transformed record by record, joined with another {@code KTable} or {@link KStream}, or
* can be re-partitioned and aggregated into a new {@code KTable}.
* <p>
* Some {@code KTable}s have an internal state (a {@link ReadOnlyKeyValueStore}) and are therefore queryable via the
* Some {@code KTable}s have an internal state (a {@link TimestampedKeyValueStore}) and are therefore queryable via the
* interactive queries API.
* For example:
* <pre>{@code
Expand All @@ -53,7 +53,7 @@
* streams.start()
* ...
* final String queryableStoreName = table.queryableStoreName(); // returns null if KTable is not queryable
* ReadOnlyKeyValueStore view = streams.store(queryableStoreName, QueryableStoreTypes.keyValueStore());
* TimestampedKeyValueStore view = streams.store(queryableStoreName, QueryableStoreTypes.keyValueStore());

@vvcephei vvcephei Dec 18, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This change doesn't look quite right. It should still be a ReadOnlyKeyValueStore, but the generics are different: ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>>. Also, if the intent is to recommend using the Timestamped variant, then folks would actually have to specify QueryableStoreTypes.timestampedKeyValueStore().

@mjsax mjsax Dec 18, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This this top level class JavaDoc, should we also explain that it it still possible to query the store as plain key-value store only? I can imagine, that not all applications are interested in the timestamp and it would be quite convenient for them to stay with the old pattern (note, that the old pattern is not deprecated and still totally valid to use).

Thoughts?

@miroswan miroswan Dec 18, 2019 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@vvcephei Thanks for the heads up. I've dug a little more and was able to confirm that streams.store returns a ReadOnlyKeyValueStore<K, ValueAndTimestamp<V>> when QueryableStoreTypes.timestampedKeyValueStore() is passed as the second argument. It looks like the type returned by streams.store is dependent upon the return type of the create method of that second argument, namely a TimestampedKeyValueStoreType. I'll make the necessary updates and verify that the code in the javadocs works.

@mjsax You raise a good point. We can provide documentation for the two approaches; however, if we add another QueryableStoreType similar to KeyValueStore and TimestampedKeyValueStore then we've got to make more updates in many places. To remedy this, perhaps we only provide a code example of how to query at the KTable interface documentation and remove the examples in the documentation for the methods. Otherwise, we can continue to update these code snippets in the documentation throughout. What do you think?

@mjsax mjsax Dec 19, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well. I see your point. Maintaining the docs in many different places is quite a challenge. Curious to hear what @vvcephei thinks about this question.

@mjsax mjsax Jan 9, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

\cc @vvcephei Any comments?

Btw: There seems to be an overlap with https://issues.apache.org/jira/browse/KAFKA-9290 -- might be worth to resolve both issue with a single PR?

* view.get(key);

@mjsax mjsax Feb 12, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: can you remove the blanks? Those would be rendered:

    // with blanks code would be intendet
    view.get(key)

// but we don't want indention
view.get(key);

This applies to all <pre>{@code... sections) -- it was not introduced by you but we should fix it.

*}</pre>
*<p>
Expand Down Expand Up @@ -130,11 +130,11 @@ public interface KTable<K, V> {
* Furthermore, for each record that gets dropped (i.e., does not satisfy the given predicate) a tombstone record
* is forwarded.
* <p>
* To query the local {@link KeyValueStore} it must be obtained via
* To query the local {@link TimestampedKeyValueStore} it must be obtained via
* {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
* <pre>{@code
* KafkaStreams streams = ... // filtering words
* ReadOnlyKeyValueStore<K,V> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, V>keyValueStore());
* TimestampedKeyValueStore<K,V> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, V>keyValueStore());
* K key = "some-word";
* V valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
* }</pre>
Expand Down Expand Up @@ -169,11 +169,11 @@ KTable<K, V> filter(final Predicate<? super K, ? super V> predicate,
* Furthermore, for each record that gets dropped (i.e., does not satisfy the given predicate) a tombstone record
* is forwarded.
* <p>
* To query the local {@link KeyValueStore} it must be obtained via
* To query the local {@link TimestampedKeyValueStore} it must be obtained via
* {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
* <pre>{@code
* KafkaStreams streams = ... // filtering words
* ReadOnlyKeyValueStore<K,V> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, V>keyValueStore());
* TimestampedKeyValueStore<K,V> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, V>keyValueStore());
* K key = "some-word";
* V valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
* }</pre>
Expand Down Expand Up @@ -255,11 +255,11 @@ KTable<K, V> filter(final Predicate<? super K, ? super V> predicate,
* Furthermore, for each record that gets dropped (i.e., does satisfy the given predicate) a tombstone record is
* forwarded.
* <p>
* To query the local {@link KeyValueStore} it must be obtained via
* To query the local {@link TimestampedKeyValueStore} it must be obtained via
* {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
* <pre>{@code
* KafkaStreams streams = ... // filtering words
* ReadOnlyKeyValueStore<K,V> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, V>keyValueStore());
* TimestampedKeyValueStore<K,V> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, V>keyValueStore());
* K key = "some-word";
* V valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
* }</pre>
Expand Down Expand Up @@ -293,11 +293,11 @@ KTable<K, V> filterNot(final Predicate<? super K, ? super V> predicate,
* Furthermore, for each record that gets dropped (i.e., does satisfy the given predicate) a tombstone record is
* forwarded.
* <p>
* To query the local {@link KeyValueStore} it must be obtained via
* To query the local {@link TimestampedKeyValueStore} it must be obtained via
* {@link KafkaStreams#store(String, QueryableStoreType) KafkaStreams#store(...)}:
* <pre>{@code
* KafkaStreams streams = ... // filtering words
* ReadOnlyKeyValueStore<K,V> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, V>keyValueStore());
* TimestampedKeyValueStore<K,V> localStore = streams.store(queryableStoreName, QueryableStoreTypes.<K, V>keyValueStore());
* K key = "some-word";
* V valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
* }</pre>
Expand Down