ableegoldman commented on a change in pull request #9139:
URL: https://github.com/apache/kafka/pull/9139#discussion_r500664927
##########
File path:
streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
##########
@@ -278,6 +326,23 @@ public void shouldFetchCorrectlyAcrossSegments() {
assertFalse(results.hasNext());
}
+ @Test
+ public void shouldBackwardFetchCorrectlyAcrossSegments() {
+ final Windowed<Bytes> a1 = new Windowed<>(keyA, new
SessionWindow(SEGMENT_INTERVAL * 0, SEGMENT_INTERVAL * 0));
+ final Windowed<Bytes> a2 = new Windowed<>(keyA, new
SessionWindow(SEGMENT_INTERVAL * 1, SEGMENT_INTERVAL * 1));
+ final Windowed<Bytes> a3 = new Windowed<>(keyA, new
SessionWindow(SEGMENT_INTERVAL * 2, SEGMENT_INTERVAL * 2));
+ cachingStore.put(a1, "1".getBytes());
+ cachingStore.put(a2, "2".getBytes());
+ cachingStore.flush();
+ cachingStore.put(a3, "3".getBytes());
Review comment:
Can we add a few more records that span multiple segments that don't get
flushed as well?
##########
File path:
streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
##########
@@ -301,6 +366,29 @@ public void shouldFetchRangeCorrectlyAcrossSegments() {
assertEquals(mkSet(a1, a2, a3, aa1, aa3), keys);
}
+ @Test
+ public void shouldBackwardFetchRangeCorrectlyAcrossSegments() {
+ final Windowed<Bytes> a1 = new Windowed<>(keyA, new
SessionWindow(SEGMENT_INTERVAL * 0, SEGMENT_INTERVAL * 0));
+ final Windowed<Bytes> aa1 = new Windowed<>(keyAA, new
SessionWindow(SEGMENT_INTERVAL * 0, SEGMENT_INTERVAL * 0));
+ final Windowed<Bytes> a2 = new Windowed<>(keyA, new
SessionWindow(SEGMENT_INTERVAL * 1, SEGMENT_INTERVAL * 1));
+ final Windowed<Bytes> a3 = new Windowed<>(keyA, new
SessionWindow(SEGMENT_INTERVAL * 2, SEGMENT_INTERVAL * 2));
+ final Windowed<Bytes> aa3 = new Windowed<>(keyAA, new
SessionWindow(SEGMENT_INTERVAL * 2, SEGMENT_INTERVAL * 2));
+ cachingStore.put(a1, "1".getBytes());
+ cachingStore.put(aa1, "1".getBytes());
+ cachingStore.put(a2, "2".getBytes());
+ cachingStore.put(a3, "3".getBytes());
+ cachingStore.put(aa3, "3".getBytes());
+
+ final KeyValueIterator<Windowed<Bytes>, byte[]> rangeResults =
+ cachingStore.backwardFindSessions(keyA, keyAA, 0, SEGMENT_INTERVAL
* 2);
+ final Set<Windowed<Bytes>> keys = new HashSet<>();
+ while (rangeResults.hasNext()) {
+ keys.add(rangeResults.next().key);
+ }
+ rangeResults.close();
+ assertEquals(mkSet(a1, a2, a3, aa1, aa3), keys);
Review comment:
We're losing the ordering check by comparing this as a set, let's use a
list (or whatever) to verify the actual order
##########
File path:
streams/src/test/java/org/apache/kafka/test/ReadOnlySessionStoreStub.java
##########
@@ -82,11 +118,15 @@ public boolean hasNext() {
public KeyValue<Windowed<K>, V> next() {
return it.next();
}
-
}
);
}
+ @Override
+ public KeyValueIterator<Windowed<K>, V> backwardFetch(K from, K to) {
+ return null;
Review comment:
I guess it probably doesn't matter since we presumably aren't using
these backward methods of the ReadOnlySessionStoreStub, but it seems like it
might result in some tricky NPEs to debug if ever someone does try to use it in
a test. If you don't feel like implementing it I think it's fine to just throw
UnsupportedOperationException and say that you'll have to implement this to use
it.
Or just copy the code from the forward direction and flip it 🤷♀️ Same goes
for all the methods in here that return null
##########
File path:
streams/src/test/java/org/apache/kafka/streams/state/internals/CachingSessionStoreTest.java
##########
@@ -456,68 +562,88 @@ public void shouldClearNamespaceCacheOnClose() {
assertEquals(0, cache.size());
}
- @Test(expected = InvalidStateStoreException.class)
+ @Test
public void shouldThrowIfTryingToFetchFromClosedCachingStore() {
cachingStore.close();
- cachingStore.fetch(keyA);
+ assertThrows(InvalidStateStoreException.class, () ->
cachingStore.fetch(keyA));
}
- @Test(expected = InvalidStateStoreException.class)
+ @Test
public void shouldThrowIfTryingToFindMergeSessionFromClosedCachingStore() {
cachingStore.close();
- cachingStore.findSessions(keyA, 0, Long.MAX_VALUE);
+ assertThrows(InvalidStateStoreException.class, () ->
cachingStore.findSessions(keyA, 0, Long.MAX_VALUE));
}
- @Test(expected = InvalidStateStoreException.class)
+ @Test
public void shouldThrowIfTryingToRemoveFromClosedCachingStore() {
cachingStore.close();
- cachingStore.remove(new Windowed<>(keyA, new SessionWindow(0, 0)));
+ assertThrows(InvalidStateStoreException.class, () ->
cachingStore.remove(new Windowed<>(keyA, new SessionWindow(0, 0))));
}
- @Test(expected = InvalidStateStoreException.class)
+ @Test
public void shouldThrowIfTryingToPutIntoClosedCachingStore() {
cachingStore.close();
- cachingStore.put(new Windowed<>(keyA, new SessionWindow(0, 0)),
"1".getBytes());
+ assertThrows(InvalidStateStoreException.class, () ->
cachingStore.put(new Windowed<>(keyA, new SessionWindow(0, 0)),
"1".getBytes()));
}
- @Test(expected = NullPointerException.class)
+ @Test
public void shouldThrowNullPointerExceptionOnFindSessionsNullKey() {
- cachingStore.findSessions(null, 1L, 2L);
+ assertThrows(NullPointerException.class, () ->
cachingStore.findSessions(null, 1L, 2L));
}
- @Test(expected = NullPointerException.class)
+ @Test
public void shouldThrowNullPointerExceptionOnFindSessionsNullFromKey() {
- cachingStore.findSessions(null, keyA, 1L, 2L);
+ assertThrows(NullPointerException.class, () ->
cachingStore.findSessions(null, keyA, 1L, 2L));
}
- @Test(expected = NullPointerException.class)
+ @Test
public void shouldThrowNullPointerExceptionOnFindSessionsNullToKey() {
- cachingStore.findSessions(keyA, null, 1L, 2L);
+ assertThrows(NullPointerException.class, () ->
cachingStore.findSessions(keyA, null, 1L, 2L));
}
- @Test(expected = NullPointerException.class)
+ @Test
public void shouldThrowNullPointerExceptionOnFetchNullFromKey() {
- cachingStore.fetch(null, keyA);
+ assertThrows(NullPointerException.class, () ->
cachingStore.fetch(null, keyA));
}
- @Test(expected = NullPointerException.class)
+ @Test
public void shouldThrowNullPointerExceptionOnFetchNullToKey() {
- cachingStore.fetch(keyA, null);
+ assertThrows(NullPointerException.class, () ->
cachingStore.fetch(keyA, null));
}
- @Test(expected = NullPointerException.class)
+ @Test
public void shouldThrowNullPointerExceptionOnFetchNullKey() {
- cachingStore.fetch(null);
+ assertThrows(NullPointerException.class, () ->
cachingStore.fetch(null));
}
- @Test(expected = NullPointerException.class)
+ @Test
public void shouldThrowNullPointerExceptionOnRemoveNullKey() {
- cachingStore.remove(null);
+ assertThrows(NullPointerException.class, () ->
cachingStore.remove(null));
}
- @Test(expected = NullPointerException.class)
+ @Test
public void shouldThrowNullPointerExceptionOnPutNullKey() {
- cachingStore.put(null, "1".getBytes());
+ assertThrows(NullPointerException.class, () -> cachingStore.put(null,
"1".getBytes()));
+ }
+
+ @Test
+ public void
shouldNotThrowInvalidBackwardRangeExceptionWithNegativeFromKey() {
Review comment:
Technically "InvalidRangeException" was just the name of the exception
that could get thrown, there is no "InvalidBackwardRangeException" that I know
of 😛 But I think the meaning is clear enough lol
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
