Versions Compared

Old Version 40

changes.mady.by.user Patrick Stuedi

Saved on

compared with

New Version Current

changes.mady.by.user John Roesler

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

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 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 timeFrom, final Instant timeTo);

    public K getKey();

    public Optional<Instant> getTimeFrom();

    public Optional<Instant> getTimeTo();
}

...

Code Block
final Instant timeTo = Instant.now();
final Instant timeFrom = timeTo.minusSeconds(60);
final WindowKeyQuery<GenericKey, ValueAndTimestamp<GenericRow>> query = WindowKeyQuery.withKeyAndWindowStartRange(key, timeFrom, timeTo);

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();

...