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 all commits
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
61 changes: 37 additions & 24 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 @@ -29,6 +29,7 @@
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 @@ -47,16 +48,28 @@
* interactive queries API.
* For example:
* <pre>{@code
* final KTable table = ...
* final KTable<String, Long> table = ...;
* ...
* final KafkaStreams streams = ...;
* streams.start()
* ...
* final String queryableStoreName = table.queryableStoreName(); // returns null if KTable is not queryable
* ReadOnlyKeyValueStore view = streams.store(queryableStoreName, QueryableStoreTypes.keyValueStore());
* ReadOnlyKeyValueStore<String, Long> view = streams.store(queryableStoreName, QueryableStoreTypes.keyValueStore());
* 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>
* }</pre>
* You may also use a {@link TimestampedKeyValueStore} when querying internal state by calling the corresponding method
* on {@link QueryableStoreTypes}.
* <pre>{@code
* final KTable<String, Long> table = ...;
* ...
* final KafkaStreams streams = ...;
* streams.start()
* ...
* final String queryableStoreName = table.queryableStoreName(); // returns null if KTable is not queryable
* ReadOnlyKeyValueStore<String, ValueAndTimestamp<Long>> view = streams.store(queryableStoreName, QueryableStoreTypes.timestampedKeyValueStore());
* view.get(key);
* }</pre>
* <p>
* Records from the source topic that have null keys are dropped.
*
* @param <K> Type of primary keys
Expand Down Expand Up @@ -130,13 +143,13 @@ 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 ReadOnlyKeyValueStore} 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());
* K key = "some-word";
* V valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
* KafkaStreams streams = ... // filtering words and assuming a KTable<String, Long>
* ReadOnlyKeyValueStore<String, ValueAndTimestamp<Long>> localStore = streams.store(queryableStoreName, QueryableStoreTypes.timestampedKeyValueStore());
* String key = "some-word"; // Key must be local (application state is shared over all running Kafka Streams instances)
* ValueAndTimestamp<Long> valueForKey = localStore.get(key);
* }</pre>
* For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
* query the value of the key on a parallel running instance of your Kafka Streams application.
Expand Down Expand Up @@ -169,13 +182,13 @@ 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 ReadOnlyKeyValueStore} 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());
* K key = "some-word";
* V valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
* KafkaStreams streams = ... // filtering words and assuming a KTable<String, Long>
* ReadOnlyKeyValueStore<String, ValueAndTimestamp<Long>> localStore = streams.store(queryableStoreName, QueryableStoreTypes.timestampedKeyValueStore());
* String key = "some-word"; // Key must be local (application state is shared over all running Kafka Streams instances)
* ValueAndTimestamp<Long> valueForKey = localStore.get(key);
* }</pre>
* For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
* query the value of the key on a parallel running instance of your Kafka Streams application.
Expand Down Expand Up @@ -255,13 +268,13 @@ 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 ReadOnlyKeyValueStore} 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());
* K key = "some-word";
* V valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
* KafkaStreams streams = ... // filtering words and assuming a KTable<String, Long>
* ReadOnlyKeyValueStore<String, ValueAndTimestamp<Long>> localStore = streams.store(queryableStoreName, QueryableStoreTypes.timestampedKeyValueStore());
* String key = "some-word"; // Key must be local (application state is shared over all running Kafka Streams instances)
* ValueAndTimestamp<Long> valueForKey = localStore.get(key);
* }</pre>
* For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
* query the value of the key on a parallel running instance of your Kafka Streams application.
Expand Down Expand Up @@ -293,13 +306,13 @@ 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 ReadOnlyKeyValueStore} 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());
* K key = "some-word";
* V valueForKey = localStore.get(key); // key must be local (application state is shared over all running Kafka Streams instances)
* KafkaStreams streams = ... // filtering words and assuming a KTable<String, Long>
* ReadOnlyKeyValueStore<String, ValueAndTimestamp<Long>> localStore = streams.store(queryableStoreName, QueryableStoreTypes.timestampedKeyValueStore());
* String key = "some-word"; // Key must be local (application state is shared over all running Kafka Streams instances)
* ValueAndTimestamp<Long> valueForKey = localStore.get(key);
* }</pre>
* For non-local keys, a custom RPC mechanism must be implemented using {@link KafkaStreams#allMetadata()} to
* query the value of the key on a parallel running instance of your Kafka Streams application.
Expand Down