Skip to content

MINOR: Remove types from caching stores #6331

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 2 commits into from
Feb 28, 2019
Merged

MINOR: Remove types from caching stores #6331

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6d4aa71
MINOR: remove types from caching stores
mjsax Feb 26, 2019
5d95689
Github comments and rebased
mjsax Feb 28, 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
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamAggregate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,7 @@ public void init(final ProcessorContext context) {
super.init(context);
metrics = (StreamsMetricsImpl) context.metrics();
store = (KeyValueStore<K, T>) context.getStateStore(storeName);
tupleForwarder = new TupleForwarder<>(store, context, new ForwardingCacheFlushListener<K, V>(context), sendOldValues);
tupleForwarder = new TupleForwarder<>(store, context, new ForwardingCacheFlushListener<>(context), sendOldValues);
}


Expand Down
2 changes: 1 addition & 1 deletion ...c/main/java/org/apache/kafka/streams/kstream/internals/KStreamSessionWindowAggregate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -93,7 +93,7 @@ public void init(final ProcessorContext context) {
lateRecordDropSensor = Sensors.lateRecordDropSensor(internalProcessorContext);

store = (SessionStore<K, Agg>) context.getStateStore(storeName);
tupleForwarder = new TupleForwarder<>(store, context, new ForwardingCacheFlushListener<K, V>(context), sendOldValues);
tupleForwarder = new TupleForwarder<>(store, context, new ForwardingCacheFlushListener<>(context), sendOldValues);
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamWindowAggregate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -89,7 +89,7 @@ public void init(final ProcessorContext context) {
lateRecordDropSensor = Sensors.lateRecordDropSensor(internalProcessorContext);

windowStore = (WindowStore<K, Agg>) context.getStateStore(storeName);
tupleForwarder = new TupleForwarder<>(windowStore, context, new ForwardingCacheFlushListener<Windowed<K>, V>(context), sendOldValues);
tupleForwarder = new TupleForwarder<>(windowStore, context, new ForwardingCacheFlushListener<>(context), sendOldValues);
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableAggregate.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,7 @@ private class KTableAggregateProcessor extends AbstractProcessor<K, Change<V>> {
public void init(final ProcessorContext context) {
super.init(context);
store = (KeyValueStore<K, T>) context.getStateStore(storeName);
tupleForwarder = new TupleForwarder<>(store, context, new ForwardingCacheFlushListener<K, V>(context), sendOldValues);
tupleForwarder = new TupleForwarder<>(store, context, new ForwardingCacheFlushListener<>(context), sendOldValues);
}

/**
Expand Down
34 changes: 6 additions & 28 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 @@ -18,7 +18,6 @@

import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.state.internals.CachedStateStore;
import org.apache.kafka.streams.state.internals.WrappedStateStore;

/**
Expand All @@ -30,45 +29,24 @@
* @param <V>
*/
class TupleForwarder<K, V> {
private final CachedStateStore cachedStateStore;
private final boolean cachingEnabled;
private final ProcessorContext context;

@SuppressWarnings("unchecked")
TupleForwarder(final StateStore store,
final ProcessorContext context,
final ForwardingCacheFlushListener flushListener,
final ForwardingCacheFlushListener<K, V> flushListener,
final boolean sendOldValues) {
this.cachedStateStore = cachedStateStore(store);
cachingEnabled = ((WrappedStateStore) store).setFlushListener(flushListener, sendOldValues);
this.context = context;
if (this.cachedStateStore != null) {
cachedStateStore.setFlushListener(flushListener, sendOldValues);
}
}

private CachedStateStore cachedStateStore(final StateStore store) {
if (store instanceof CachedStateStore) {
return (CachedStateStore) store;
} else if (store instanceof WrappedStateStore) {
StateStore wrapped = ((WrappedStateStore) store).wrapped();

while (wrapped instanceof WrappedStateStore && !(wrapped instanceof CachedStateStore)) {
Copy link
Copy Markdown
Member

@bbejeck bbejeck Feb 27, 2019

Choose a reason for hiding this comment

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

The PR replaces a while loop with a single statement, does this mean Streams is guaranteed to only ever have one level of a WrappedStateStore?

Copy link
Copy Markdown
Member Author

@mjsax mjsax Feb 27, 2019

Choose a reason for hiding this comment

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

TupleForwarder is part of the DSL and thus we know the wrapping hierarchy, and hence, this should be save. We know that the caching store is either inside metered stores, or there is no caching store.

wrapped = ((WrappedStateStore) wrapped).wrapped();
}

if (!(wrapped instanceof CachedStateStore)) {
return null;
}

return (CachedStateStore) wrapped;
}
return null;
}

public void maybeForward(final K key,
final V newValue,
final V oldValue) {
if (cachedStateStore == null) {
context.forward(key, new Change<>(newValue, oldValue));
if (cachingEnabled) {
return;
}
context.forward(key, new Change<>(newValue, oldValue));
}
}
20 changes: 12 additions & 8 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 @@ -204,7 +204,9 @@ public Cancellable schedule(final Duration interval,
return schedule(ApiUtils.validateMillisecondDuration(interval, msgPrefix), type, callback);
}

private abstract static class StateStoreReadOnlyDecorator<T extends StateStore> extends WrappedStateStore<T> {
private abstract static class StateStoreReadOnlyDecorator<T extends StateStore, K, V>
extends WrappedStateStore<T, K, V> {

static final String ERROR_MESSAGE = "Global store is read only";

private StateStoreReadOnlyDecorator(final T inner) {
Expand All @@ -229,7 +231,7 @@ public void close() {
}

private static class KeyValueStoreReadOnlyDecorator<K, V>
extends StateStoreReadOnlyDecorator<KeyValueStore<K, V>>
extends StateStoreReadOnlyDecorator<KeyValueStore<K, V>, K, V>
implements KeyValueStore<K, V> {

private KeyValueStoreReadOnlyDecorator(final KeyValueStore<K, V> inner) {
Expand Down Expand Up @@ -281,7 +283,7 @@ public V delete(final K key) {
}

private static class WindowStoreReadOnlyDecorator<K, V>
extends StateStoreReadOnlyDecorator<WindowStore<K, V>>
extends StateStoreReadOnlyDecorator<WindowStore<K, V>, K, V>
implements WindowStore<K, V> {

private WindowStoreReadOnlyDecorator(final WindowStore<K, V> inner) {
Expand Down Expand Up @@ -338,7 +340,7 @@ public KeyValueIterator<Windowed<K>, V> fetchAll(final long timeFrom,
}

private static class SessionStoreReadOnlyDecorator<K, AGG>
extends StateStoreReadOnlyDecorator<SessionStore<K, AGG>>
extends StateStoreReadOnlyDecorator<SessionStore<K, AGG>, K, AGG>
implements SessionStore<K, AGG> {

private SessionStoreReadOnlyDecorator(final SessionStore<K, AGG> inner) {
Expand Down Expand Up @@ -388,7 +390,9 @@ public KeyValueIterator<Windowed<K>, AGG> fetch(final K from,
}
}

private abstract static class StateStoreReadWriteDecorator<T extends StateStore> extends WrappedStateStore<T> {
private abstract static class StateStoreReadWriteDecorator<T extends StateStore, K, V>
extends WrappedStateStore<T, K, V> {

static final String ERROR_MESSAGE = "This method may only be called by Kafka Streams";

private StateStoreReadWriteDecorator(final T inner) {
Expand All @@ -408,7 +412,7 @@ public void close() {
}

private static class KeyValueStoreReadWriteDecorator<K, V>
extends StateStoreReadWriteDecorator<KeyValueStore<K, V>>
extends StateStoreReadWriteDecorator<KeyValueStore<K, V>, K, V>
implements KeyValueStore<K, V> {

private KeyValueStoreReadWriteDecorator(final KeyValueStore<K, V> inner) {
Expand Down Expand Up @@ -460,7 +464,7 @@ public V delete(final K key) {
}

private static class WindowStoreReadWriteDecorator<K, V>
extends StateStoreReadWriteDecorator<WindowStore<K, V>>
extends StateStoreReadWriteDecorator<WindowStore<K, V>, K, V>
implements WindowStore<K, V> {

private WindowStoreReadWriteDecorator(final WindowStore<K, V> inner) {
Expand Down Expand Up @@ -517,7 +521,7 @@ public KeyValueIterator<Windowed<K>, V> fetchAll(final long timeFrom,
}

private static class SessionStoreReadWriteDecorator<K, AGG>
extends StateStoreReadWriteDecorator<SessionStore<K, AGG>>
extends StateStoreReadWriteDecorator<SessionStore<K, AGG>, K, AGG>
implements SessionStore<K, AGG> {

private SessionStoreReadWriteDecorator(final SessionStore<K, AGG> inner) {
Expand Down
4 changes: 2 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/CachedStateStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,6 @@ public interface CachedStateStore<K, V> {
* @param listener
* @param sendOldValues
*/
void setFlushListener(final CacheFlushListener<K, V> listener,
final boolean sendOldValues);
boolean setFlushListener(final CacheFlushListener<K, V> listener,
Copy link
Copy Markdown
Member Author

@mjsax mjsax Feb 27, 2019

Choose a reason for hiding this comment

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

We return true if the listener was registered, ie, if the "metered stores" wraps a "caching store".

Copy link
Copy Markdown
Contributor

@vvcephei vvcephei Feb 27, 2019

Choose a reason for hiding this comment

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

Clever! I tried to do this change a little while ago and got hung up on this point.

Copy link
Copy Markdown
Member Author

@mjsax mjsax Feb 27, 2019

Choose a reason for hiding this comment

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

Thanks. blushing

final boolean sendOldValues);
}
Loading