Skip to content

KAFKA-6970: All standard state stores guarded with read only wrapper #6016

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
mjsax merged 7 commits into from
Dec 11, 2018
Merged

KAFKA-6970: All standard state stores guarded with read only wrapper #6016

Show file tree
Hide file tree
Changes from 6 commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
70f7067
KAFKA-6970: All standard state stores from ProcessContext are guarded…
nizhikov Dec 8, 2018
6ab1684
KAFKA-6970: Reworked to guard local state store with ReadWriteDecorator.
nizhikov Dec 9, 2018
84c0e80
KAFKA-6970: Fixing test.
nizhikov Dec 9, 2018
54c1b9c
KAFKA-6970: Tests fixed. Also fixing wrap classes for StateStore
nizhikov Dec 9, 2018
970e612
KAFKA-6970: Style fix.
nizhikov Dec 9, 2018
ba977d9
KAFKA-6970: Code review fixes.
nizhikov Dec 10, 2018
6e2ba43
KAFKA-6970: Code review fixes.
nizhikov Dec 10, 2018
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
14 changes: 11 additions & 3 deletions streams/src/main/java/org/apache/kafka/streams/kstream/internals/TupleForwarder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,9 +50,17 @@ class TupleForwarder<K, V> {
private CachedStateStore cachedStateStore(final StateStore store) {
if (store instanceof CachedStateStore) {
return (CachedStateStore) store;
} else if (store instanceof WrappedStateStore
&& ((WrappedStateStore) store).wrappedStore() instanceof CachedStateStore) {
return (CachedStateStore) ((WrappedStateStore) store).wrappedStore();
} else if (store instanceof WrappedStateStore) {
StateStore wrapped = ((WrappedStateStore) store).wrappedStore();

while (wrapped instanceof WrappedStateStore && !(wrapped instanceof CachedStateStore)) {
wrapped = ((WrappedStateStore) wrapped).wrappedStore();
}

if (!(wrapped instanceof CachedStateStore))
return null;
Comment thread
nizhikov marked this conversation as resolved.

return (CachedStateStore) wrapped;
}
return null;
}
Expand Down
232 changes: 190 additions & 42 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.processor.internals;

import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.internals.ApiUtils;
Expand All @@ -37,6 +38,7 @@

import java.time.Duration;
import java.util.List;
import org.apache.kafka.streams.state.internals.WrappedStateStore.AbstractStateStore;

import static org.apache.kafka.streams.internals.ApiUtils.prepareMillisCheckFailMsgPrefix;

Expand Down Expand Up @@ -102,7 +104,16 @@ public StateStore getStateStore(final String name) {
"please file a bug report at https://issues.apache.org/jira/projects/KAFKA.");
}

return stateManager.getStore(name);
final StateStore store = stateManager.getStore(name);
if (store instanceof KeyValueStore) {
return new KeyValueStoreReadWriteDecorator((KeyValueStore) store);
} else if (store instanceof WindowStore) {
return new WindowStoreReadWriteDecorator((WindowStore) store);
} else if (store instanceof SessionStore) {
return new SessionStoreReadWriteDecorator((SessionStore) store);
}

return store;
}

@SuppressWarnings("unchecked")
Expand Down Expand Up @@ -196,23 +207,16 @@ public long streamTime() {
return streamTimeSupplier.get();
}

private abstract static class StateStoreReadOnlyDecorator<T extends StateStore> implements StateStore {
private abstract static class StateStoreReadOnlyDecorator<T extends StateStore> extends AbstractStateStore {
static final String ERROR_MESSAGE = "Global store is read only";

final T underlying;

StateStoreReadOnlyDecorator(final T underlying) {
this.underlying = underlying;
}

@Override
public String name() {
return underlying.name();
StateStoreReadOnlyDecorator(final T inner) {
super(inner);
}

@Override
public void init(final ProcessorContext context, final StateStore root) {
Comment thread
nizhikov marked this conversation as resolved.
underlying.init(context, root);
@SuppressWarnings("unchecked")
T getInner() {
return (T) wrappedStore();
}

@Override
Expand All @@ -221,44 +225,39 @@ public void flush() {
}

@Override
public void close() {
underlying.close();
}

@Override
public boolean persistent() {
return underlying.persistent();
public void init(final ProcessorContext context, final StateStore root) {
throw new UnsupportedOperationException(ERROR_MESSAGE);
}

@Override
public boolean isOpen() {
return underlying.isOpen();
public void close() {
throw new UnsupportedOperationException(ERROR_MESSAGE);
}
}

private static class KeyValueStoreReadOnlyDecorator<K, V> extends StateStoreReadOnlyDecorator<KeyValueStore<K, V>> implements KeyValueStore<K, V> {
KeyValueStoreReadOnlyDecorator(final KeyValueStore<K, V> underlying) {
super(underlying);
KeyValueStoreReadOnlyDecorator(final KeyValueStore<K, V> inner) {
super(inner);
}

@Override
public V get(final K key) {
return underlying.get(key);
return getInner().get(key);
}

@Override
public KeyValueIterator<K, V> range(final K from, final K to) {
return underlying.range(from, to);
return getInner().range(from, to);
}

@Override
public KeyValueIterator<K, V> all() {
return underlying.all();
return getInner().all();
}

@Override
public long approximateNumEntries() {
return underlying.approximateNumEntries();
return getInner().approximateNumEntries();
}

@Override
Expand All @@ -283,8 +282,8 @@ public V delete(final K key) {
}

private static class WindowStoreReadOnlyDecorator<K, V> extends StateStoreReadOnlyDecorator<WindowStore<K, V>> implements WindowStore<K, V> {
WindowStoreReadOnlyDecorator(final WindowStore<K, V> underlying) {
super(underlying);
WindowStoreReadOnlyDecorator(final WindowStore<K, V> inner) {
super(inner);
}

@Override
Expand All @@ -299,46 +298,46 @@ public void put(final K key, final V value, final long windowStartTimestamp) {

@Override
public V fetch(final K key, final long time) {
return underlying.fetch(key, time);
return getInner().fetch(key, time);
}

@Deprecated
@Override
public WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo) {
return underlying.fetch(key, timeFrom, timeTo);
return getInner().fetch(key, timeFrom, timeTo);
}

@Deprecated
@Override
public KeyValueIterator<Windowed<K>, V> fetch(final K from, final K to, final long timeFrom, final long timeTo) {
return underlying.fetch(from, to, timeFrom, timeTo);
return getInner().fetch(from, to, timeFrom, timeTo);
}

@Override
public KeyValueIterator<Windowed<K>, V> all() {
return underlying.all();
return getInner().all();
}

@Deprecated
@Override
public KeyValueIterator<Windowed<K>, V> fetchAll(final long timeFrom, final long timeTo) {
return underlying.fetchAll(timeFrom, timeTo);
return getInner().fetchAll(timeFrom, timeTo);
}
}

private static class SessionStoreReadOnlyDecorator<K, AGG> extends StateStoreReadOnlyDecorator<SessionStore<K, AGG>> implements SessionStore<K, AGG> {
SessionStoreReadOnlyDecorator(final SessionStore<K, AGG> underlying) {
super(underlying);
SessionStoreReadOnlyDecorator(final SessionStore<K, AGG> inner) {
super(inner);
}

@Override
public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) {
return underlying.findSessions(key, earliestSessionEndTime, latestSessionStartTime);
return getInner().findSessions(key, earliestSessionEndTime, latestSessionStartTime);
}

@Override
public KeyValueIterator<Windowed<K>, AGG> findSessions(final K keyFrom, final K keyTo, final long earliestSessionEndTime, final long latestSessionStartTime) {
return underlying.findSessions(keyFrom, keyTo, earliestSessionEndTime, latestSessionStartTime);
return getInner().findSessions(keyFrom, keyTo, earliestSessionEndTime, latestSessionStartTime);
}

@Override
Expand All @@ -353,12 +352,161 @@ public void put(final Windowed<K> sessionKey, final AGG aggregate) {

@Override
public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) {
return underlying.fetch(key);
return getInner().fetch(key);
}

@Override
public KeyValueIterator<Windowed<K>, AGG> fetch(final K from, final K to) {

@vvcephei vvcephei Dec 10, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Good catch!

return getInner().fetch(from, to);
}
}

private abstract static class StateStoreReadWriteDecorator<T extends StateStore> extends AbstractStateStore {
static final String ERROR_MESSAGE = "This method may only be called by Kafka Streams";

StateStoreReadWriteDecorator(final T inner) {
super(inner);
}

@SuppressWarnings("unchecked")
T wrapped() {
return (T) super.wrappedStore();
}

@Override
public void init(final ProcessorContext context, final StateStore root) {
throw new UnsupportedOperationException(ERROR_MESSAGE);
}

@Override
public void close() {
throw new UnsupportedOperationException(ERROR_MESSAGE);
}
}

private static class KeyValueStoreReadWriteDecorator<K, V> extends StateStoreReadWriteDecorator<KeyValueStore<K, V>> implements KeyValueStore<K, V> {
KeyValueStoreReadWriteDecorator(final KeyValueStore<K, V> inner) {
super(inner);
}

@Override
public V get(final K key) {
return wrapped().get(key);
}

@Override
public KeyValueIterator<K, V> range(final K from, final K to) {
return wrapped().range(from, to);
}

@Override
public KeyValueIterator<K, V> all() {
return wrapped().all();
}

@Override
public long approximateNumEntries() {
return wrapped().approximateNumEntries();
}

@Override
public void put(final K key, final V value) {
wrapped().put(key, value);
}

@Override
public V putIfAbsent(final K key, final V value) {
return wrapped().putIfAbsent(key, value);
}

@Override
public void putAll(final List<KeyValue<K, V>> entries) {
wrapped().putAll(entries);
}

@Override
public V delete(final K key) {
return wrapped().delete(key);
}
}

private static class WindowStoreReadWriteDecorator<K, V> extends StateStoreReadWriteDecorator<WindowStore<K, V>> implements WindowStore<K, V> {
WindowStoreReadWriteDecorator(final WindowStore<K, V> inner) {
super(inner);
}

@Override
public void put(final K key, final V value) {
wrapped().put(key, value);
}

@Override
public void put(final K key, final V value, final long windowStartTimestamp) {
wrapped().put(key, value, windowStartTimestamp);
}

@Override
public V fetch(final K key, final long time) {
return wrapped().fetch(key, time);
}

@Deprecated
@Override
public WindowStoreIterator<V> fetch(final K key, final long timeFrom, final long timeTo) {
return wrapped().fetch(key, timeFrom, timeTo);
}

@Deprecated
@Override
public KeyValueIterator<Windowed<K>, V> fetch(final K from, final K to, final long timeFrom, final long timeTo) {
return wrapped().fetch(from, to, timeFrom, timeTo);
}

@Override
public KeyValueIterator<Windowed<K>, V> all() {
return wrapped().all();
}

@Deprecated
@Override
public KeyValueIterator<Windowed<K>, V> fetchAll(final long timeFrom, final long timeTo) {
return wrapped().fetchAll(timeFrom, timeTo);
}
}

private static class SessionStoreReadWriteDecorator<K, AGG> extends StateStoreReadWriteDecorator<SessionStore<K, AGG>> implements SessionStore<K, AGG> {
SessionStoreReadWriteDecorator(final SessionStore<K, AGG> inner) {
super(inner);
}

@Override
public KeyValueIterator<Windowed<K>, AGG> findSessions(final K key, final long earliestSessionEndTime, final long latestSessionStartTime) {
return wrapped().findSessions(key, earliestSessionEndTime, latestSessionStartTime);
}

@Override
public KeyValueIterator<Windowed<K>, AGG> findSessions(final K keyFrom, final K keyTo, final long earliestSessionEndTime, final long latestSessionStartTime) {
return wrapped().findSessions(keyFrom, keyTo, earliestSessionEndTime, latestSessionStartTime);
}

@Override
public void remove(final Windowed<K> sessionKey) {
wrapped().remove(sessionKey);
}

@Override
public void put(final Windowed<K> sessionKey, final AGG aggregate) {
wrapped().put(sessionKey, aggregate);
}

@Override
public KeyValueIterator<Windowed<K>, AGG> fetch(final K key) {
return wrapped().fetch(key);
}

@Override
public KeyValueIterator<Windowed<K>, AGG> fetch(final K from, final K to) {
return underlying.fetch(from, to);
return wrapped().fetch(from, to);
}
}
}
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/state/internals/WrappedStateStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,7 @@ public interface WrappedStateStore extends StateStore {
abstract class AbstractStateStore implements WrappedStateStore {
final StateStore innerState;

AbstractStateStore(final StateStore inner) {
protected AbstractStateStore(final StateStore inner) {
this.innerState = inner;
}

Expand Down
Loading