Skip to content

KAFKA-9929: Support backward iterator on SessionStore #9139

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

Merged
vvcephei merged 15 commits into from
Oct 8, 2020
Merged

KAFKA-9929: Support backward iterator on SessionStore #9139

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
9d4327c
key/value reverse operation
jeqo Aug 5, 2020
6761c71
window backward operations
jeqo Aug 5, 2020
000c7b7
session backward operations
jeqo Aug 5, 2020
1811b57
improve time range comments
jeqo Aug 11, 2020
8c26756
fix bytes range validator not needed
jeqo Aug 11, 2020
e0d160b
fix cache iterator
jeqo Aug 11, 2020
ae7e3ed
fix session store to support backward queries
jeqo Sep 7, 2020
64a95c0
improve format based on feedback
jeqo Oct 6, 2020
a875cf1
test caching store across segments
jeqo Oct 6, 2020
76bb9a2
improve testing names and scenarios
jeqo Oct 7, 2020
e49f258
expand caching store tests per backing store
jeqo Oct 7, 2020
b856242
fix formatting
Oct 8, 2020
1932413
fix formatting
Oct 8, 2020
70484e4
Merge branch 'trunk' into backward-sessionstore
Oct 8, 2020
14fce52
fix conflict with trunk
Oct 8, 2020
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
130 changes: 126 additions & 4 deletions streams/src/main/java/org/apache/kafka/streams/state/ReadOnlySessionStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,31 +28,153 @@
* @param <AGG> the aggregated value type
*/
public interface ReadOnlySessionStore<K, AGG> {

/**
* Retrieve all aggregated sessions for the provided key.
* Fetch any sessions with the matching key and the sessions end is &ge; earliestSessionEndTime and the sessions
* start is &le; latestSessionStartTime iterating from earliest to latest.
* <p>
* This iterator must be closed after use.
*
* @param key the key to return sessions for
* @param earliestSessionEndTime the end timestamp of the earliest session to search for, where iteration starts.
* @param latestSessionStartTime the end timestamp of the latest session to search for, where iteration ends.
* @return iterator of sessions with the matching key and aggregated values, from earliest to latest session time.
* @throws NullPointerException If null is used for key.
*/
default KeyValueIterator<Windowed<K>, AGG> findSessions(final K key,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");
}

/**
* Fetch any sessions with the matching key and the sessions end is &ge; earliestSessionEndTime and the sessions
* start is &le; latestSessionStartTime iterating from latest to earliest.
* <p>
* This iterator must be closed after use.
*
* @param key the key to return sessions for
* @param earliestSessionEndTime the end timestamp of the earliest session to search for, where iteration ends.
* @param latestSessionStartTime the end timestamp of the latest session to search for, where iteration starts.
* @return backward iterator of sessions with the matching key and aggregated values, from latest to earliest session time.
* @throws NullPointerException If null is used for key.
*/
default KeyValueIterator<Windowed<K>, AGG> backwardFindSessions(final K key,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");
}

/**
* Fetch any sessions in the given range of keys and the sessions end is &ge; earliestSessionEndTime and the sessions
* start is &le; latestSessionStartTime iterating from earliest to latest.
* <p>
* This iterator must be closed after use.
*
* @param keyFrom The first key that could be in the range
* @param keyTo The last key that could be in the range
* @param earliestSessionEndTime the end timestamp of the earliest session to search for, where iteration starts.
* @param latestSessionStartTime the end timestamp of the latest session to search for, where iteration ends.
* @return iterator of sessions with the matching keys and aggregated values, from earliest to latest session time.
* @throws NullPointerException If null is used for any key.
*/
default KeyValueIterator<Windowed<K>, AGG> findSessions(final K keyFrom,
final K keyTo,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");
}


/**
* Fetch any sessions in the given range of keys and the sessions end is &ge; earliestSessionEndTime and the sessions
* start is &le; latestSessionStartTime iterating from latest to earliest.
* <p>
* This iterator must be closed after use.
*
* @param keyFrom The first key that could be in the range
* @param keyTo The last key that could be in the range
* @param earliestSessionEndTime the end timestamp of the earliest session to search for, where iteration ends.
* @param latestSessionStartTime the end timestamp of the latest session to search for, where iteration starts.
* @return backward iterator of sessions with the matching keys and aggregated values, from latest to earliest session time.
* @throws NullPointerException If null is used for any key.
*/
default KeyValueIterator<Windowed<K>, AGG> backwardFindSessions(final K keyFrom,
final K keyTo,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");
}

/**
* Get the value of key from a single session.
*
* @param key the key to fetch
* @param startTime start timestamp of the session
* @param endTime end timestamp of the session
* @return The value or {@code null} if no session associated with the key can be found
* @throws NullPointerException If {@code null} is used for any key.
*/
default AGG fetchSession(final K key, final long startTime, final long endTime) {
throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");
}

/**
* Retrieve all aggregated sessions for the provided key.
* This iterator must be closed after use.
* <p>
* For each key, the iterator guarantees ordering of sessions, starting from the oldest/earliest
* available session to the newest/latest session.
*
* @param key record key to find aggregated session values for
* @return KeyValueIterator containing all sessions for the provided key.
* @return KeyValueIterator containing all sessions for the provided key, from oldest to newest session.
* @throws NullPointerException If null is used for key.
*
*/
KeyValueIterator<Windowed<K>, AGG> fetch(final K key);

/**
* Retrieve all aggregated sessions for the given range of keys.
* Retrieve all aggregated sessions for the provided key.
* This iterator must be closed after use.
* <p>
* For each key, the iterator guarantees ordering of sessions, starting from the newest/latest
* available session to the oldest/earliest session.
*
* @param key record key to find aggregated session values for
* @return backward KeyValueIterator containing all sessions for the provided key, from newest to oldest session.
* @throws NullPointerException If null is used for key.
*/
default KeyValueIterator<Windowed<K>, AGG> backwardFetch(final K key) {
throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");
}

/**
* Retrieve all aggregated sessions for the given range of keys.
* This iterator must be closed after use.
* <p>
* For each key, the iterator guarantees ordering of sessions, starting from the oldest/earliest
* available session to the newest/latest session.
*
* @param from first key in the range to find aggregated session values for
* @param to last key in the range to find aggregated session values for
* @return KeyValueIterator containing all sessions for the provided key.
* @return KeyValueIterator containing all sessions for the provided key, from oldest to newest session.
* @throws NullPointerException If null is used for any of the keys.
*/
KeyValueIterator<Windowed<K>, AGG> fetch(final K from, final K to);

/**
* Retrieve all aggregated sessions for the given range of keys.
* This iterator must be closed after use.
* <p>
* For each key, the iterator guarantees ordering of sessions, starting from the newest/latest
* available session to the oldest/earliest session.
*
* @param from first key in the range to find aggregated session values for
* @param to last key in the range to find aggregated session values for
* @return backward KeyValueIterator containing all sessions for the provided key, from newest to oldest session.
* @throws NullPointerException If null is used for any of the keys.
*/
default KeyValueIterator<Windowed<K>, AGG> backwardFetch(final K from, final K to) {
throw new UnsupportedOperationException("This API is not supported by this implementation of ReadOnlySessionStore.");
}
}
40 changes: 0 additions & 40 deletions streams/src/main/java/org/apache/kafka/streams/state/SessionStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,46 +34,6 @@
*/
public interface SessionStore<K, AGG> extends StateStore, ReadOnlySessionStore<K, AGG> {

/**
* Fetch any sessions with the matching key and the sessions end is &ge; earliestSessionEndTime and the sessions
* start is &le; latestSessionStartTime
*
* This iterator must be closed after use.
*
* @param key the key to return sessions for
* @param earliestSessionEndTime the end timestamp of the earliest session to search for
* @param latestSessionStartTime the end timestamp of the latest session to search for
* @return iterator of sessions with the matching key and aggregated values
* @throws NullPointerException If null is used for key.
*/
KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime);

/**
* Fetch any sessions in the given range of keys and the sessions end is &ge; earliestSessionEndTime and the sessions
* start is &le; latestSessionStartTime
*
* This iterator must be closed after use.
*
* @param keyFrom The first key that could be in the range
* @param keyTo The last key that could be in the range
* @param earliestSessionEndTime the end timestamp of the earliest session to search for
* @param latestSessionStartTime the end timestamp of the latest session to search for
* @return iterator of sessions with the matching keys and aggregated values
* @throws NullPointerException If null is used for any key.
*/
KeyValueIterator<Windowed<K>, AGG> findSessions(final K keyFrom, final K keyTo, final long earliestSessionEndTime, final long latestSessionStartTime);

/**
* Get the value of key from a single session.
*
* @param key the key to fetch
* @param startTime start timestamp of the session
* @param endTime end timestamp of the session
* @return The value or {@code null} if no session associated with the key can be found
* @throws NullPointerException If {@code null} is used for any key.
*/
AGG fetchSession(final K key, final long startTime, final long endTime);

/**
* Remove the session aggregated with provided {@link Windowed} key from the store
* @param sessionKey key of the session to remove
Expand Down
Loading