...
This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Status
Current state: "Under Discussion"Adopted
Discussion thread: here
JIRA: KAFKA-13494
...
Code Block |
---|
@Evolving public class WindowKeyQuery<K, V> implements Query<WindowStoreIterator<V>> { public static <K, V> WindowKeyQuery<K, V> withKeyAndWindowStartRange(final K key, final Instant startTimetimeFrom, final Instant endTimetimeTo); public K getKey(); public Optional<Instant> getStartTimegetTimeFrom(); public Optional<Instant> getEndTimegetTimeTo(); } |
The WindowRangeQuery class:
Code Block |
---|
@Evolving public class WindowRangeQuery<K, V> implements Query<KeyValueIterator<Windowed<K>, V>> { public static <K, V> WindowRangeQuery<K, V> withKey(final K key); public static <K, V> WindowRangeQuery<K, V> withWindowStartRange(final Instant earliestWindowStartTimetimeFrom, final Instant latestWindowStartTimetimeTo); public K getKey(); public Optional<Instant> getStartTimegetTimeFrom(); public Optional<Instant> getEndTimegetTimeTo(); } |
Examples
The following example illustrates the use of the WindowKeyQuey class.
Code Block |
---|
final Instant uppertimeTo = Instant.now(); final Instant lowertimeFrom = uppertimeTo.minusSeconds(60); final WindowKeyQuery<GenericKey, ValueAndTimestamp<GenericRow>> query = WindowKeyQuery.withKeyAndWindowStartRange(key, lowertimeFrom, uppertimeTo); final StateQueryRequest<WindowStoreIterator<ValueAndTimestamp<GenericRow>>> request = inStore("rocksdb-window-store").withQuery(query); final StateQueryResult<WindowStoreIterator<ValueAndTimestamp<GenericRow>>> result = stateStore.getKafkaStreams()kafkaStream.query(request); final WindowStoreIterator<ValueAndTimestamp<GenericRow>> iterator = result.getGlobalResult().getResult(); |
The following example illustrates the use of the WindowQuery class to query a window store.
Code Block |
---|
final Instant uppertimeTo = Instant.now(); final Instant lowertimeFrom = uppertimeTo.minusSeconds(60); final WindowRangeQuery<GenericKey, ValueAndTimestamp<GenericRow>> query = WindowRangeQuery.withWindowRangewithWindowStartRange(lowertimeFrom, uppertimeTo); final StateQueryRequest<KeyValueIterator<Windowed<GenericKey>, ValueAndTimestamp<GenericRow>>> request = inStore("inmemory-window-store").withQuery(query); final StateQueryResult<KeyValueIterator<Windowed<GenericKey>, ValueAndTimestamp<GenericRow>>> result = stateStore.getKafkaStreams()kafkaStreams.query(request); final KeyValueIterator<Windowed<GenericKey>, ValueAndTimestamp<GenericRow>> iterator = result.getGlobalResult().getResult(); |
...