Skip to content

[MINOR] Guard against crashing on invalid key range queries #6521

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
bbejeck merged 10 commits into from
Apr 10, 2019
Merged

[MINOR] Guard against crashing on invalid key range queries #6521

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
1167df5
Guard against crashing on invalid key range queries in CachingXXXStores
ableegoldman Mar 29, 2019
09ace6b
Added guards for range queries in in-memory stores
ableegoldman Mar 29, 2019
ac27e85
Changed DEBUG log level to WARN
ableegoldman Apr 4, 2019
b961d0a
Addressing comments from Bruno's review
ableegoldman Apr 5, 2019
cef3de5
Added unit tests, guard range in MemoryNavigableLRUCache, expanded er…
ableegoldman Apr 5, 2019
f2a556b
Use LogAppender to catch and check logged messages in unit tests
ableegoldman Apr 5, 2019
b2deb14
Extend check & log error message to RocksDB stores as well
ableegoldman Apr 8, 2019
61e86c8
Verify empty iterator returned when fetching from negative starting k…
ableegoldman Apr 8, 2019
bf38b11
Unit tests for key range fetch to match single-key fetch in all stores
ableegoldman Apr 8, 2019
892b1e9
Fix checkstyle
ableegoldman Apr 8, 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
10 changes: 10 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,11 +30,15 @@
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class CachingKeyValueStore
extends WrappedStateStore<KeyValueStore<Bytes, byte[]>, byte[], byte[]>
implements KeyValueStore<Bytes, byte[]>, CachedStateStore<byte[], byte[]> {

private static final Logger LOG = LoggerFactory.getLogger(CachingKeyValueStore.class);

private CacheFlushListener<byte[], byte[]> flushListener;
private boolean sendOldValues;
private String cacheName;
Expand Down Expand Up @@ -228,6 +232,12 @@ private byte[] getInternal(final Bytes key) {
@Override
public KeyValueIterator<Bytes, byte[]> range(final Bytes from,
final Bytes to) {
// Make sure this is a valid query

@cadonna cadonna Apr 5, 2019

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: I would remove the comment here (and in all occurrences below), because the code itself is clear enough about what it does. Maybe rename from and to to fromKey and toKey (or similar) to make it even more clearer. Renaming would also apply to some of the changes below.

@ableegoldman ableegoldman Apr 5, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ack

if (from.compareTo(to) > 0) {
LOG.debug("Returning empty iterator for range query with invalid range: keyFrom > keyTo.");

@bbejeck bbejeck Apr 4, 2019

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: since this represents an invalid range maybe this could be a WARN?

@ableegoldman ableegoldman Apr 4, 2019

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ack, good point

return KeyValueIterators.emptyIterator();
}

validateStoreOpen();
final KeyValueIterator<Bytes, byte[]> storeIterator = wrapped().range(from, to);
final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(cacheName, from, to);
Expand Down
10 changes: 10 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,15 @@
import org.apache.kafka.streams.state.SessionStore;

import java.util.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class CachingSessionStore
extends WrappedStateStore<SessionStore<Bytes, byte[]>, byte[], byte[]>
implements SessionStore<Bytes, byte[]>, CachedStateStore<byte[], byte[]> {

private static final Logger LOG = LoggerFactory.getLogger(CachingSessionStore.class);

private final SessionKeySchema keySchema;
private final SegmentedCacheFunction cacheFunction;
private String cacheName;
Expand Down Expand Up @@ -153,6 +157,12 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final Bytes keyFro
final Bytes keyTo,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
// Make sure this is a valid query
if (keyFrom.compareTo(keyTo) > 0) {
LOG.debug("Returning empty iterator for findSessions call with invalid range: keyFrom > keyTo.");

@bbejeck bbejeck Apr 4, 2019

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.

ditto

return KeyValueIterators.emptyIterator();
}

validateStoreOpen();

final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRange(keyFrom, earliestSessionEndTime));
Expand Down
10 changes: 10 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,11 +28,15 @@
import org.apache.kafka.streams.state.StateSerdes;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

class CachingWindowStore
extends WrappedStateStore<WindowStore<Bytes, byte[]>, byte[], byte[]>
implements WindowStore<Bytes, byte[]>, CachedStateStore<byte[], byte[]> {

private static final Logger LOG = LoggerFactory.getLogger(CachingWindowStore.class);

private final long windowSize;
private final SegmentedBytesStore.KeySchema keySchema = new WindowKeySchema();

Expand Down Expand Up @@ -196,6 +200,12 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes from,
final Bytes to,
final long timeFrom,
final long timeTo) {
// Make sure this is a valid query
if (from.compareTo(to) > 0) {
LOG.debug("Returning empty iterator for fetch with invalid range: keyFrom > keyTo.");
return KeyValueIterators.emptyIterator();

@bbejeck bbejeck Apr 4, 2019

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.

ditto here and below

}

// since this function may not access the underlying inner store, we need to validate
// if store is open outside as well.
validateStoreOpen();
Expand Down
10 changes: 10 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,12 +28,16 @@

import java.util.Iterator;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class InMemoryKeyValueStore implements KeyValueStore<Bytes, byte[]> {
private final String name;
private final ConcurrentNavigableMap<Bytes, byte[]> map;
private volatile boolean open = false;

private static final Logger LOG = LoggerFactory.getLogger(InMemoryKeyValueStore.class);

public InMemoryKeyValueStore(final String name) {
this.name = name;

Expand Down Expand Up @@ -111,6 +115,12 @@ public byte[] delete(final Bytes key) {

@Override
public KeyValueIterator<Bytes, byte[]> range(final Bytes from, final Bytes to) {
// Make sure this is a valid query
if (from.compareTo(to) > 0) {
LOG.debug("Returning empty iterator for range query with invalid range: keyFrom > keyTo.");
return KeyValueIterators.emptyIterator();
}

return new DelegatingPeekingKeyValueIterator<>(
name,
new InMemoryKeyValueIterator(this.map.subMap(from, true, to, true).entrySet().iterator()));
Expand Down
6 changes: 6 additions & 0 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,6 +185,12 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes from,
final long timeTo) {
removeExpiredSegments();

// Make sure this is a valid query
if (from.compareTo(to) > 0) {
LOG.debug("Returning empty iterator for fetch with invalid range: keyFrom > keyTo.");
return KeyValueIterators.emptyIterator();
}

// add one b/c records expire exactly retentionPeriod ms after created
final long minTime = Math.max(timeFrom, this.observedStreamTime - this.retentionPeriod + 1);

Expand Down