Skip to content

KAFKA-3522: Add TimestampedWindowStore builder/runtime classes #6173

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 3 commits into from
Mar 8, 2019
Merged

KAFKA-3522: Add TimestampedWindowStore builder/runtime classes #6173

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b03a811
KAFKA-3522: Add TimestampedWindowStore builder/runtime classes
mjsax Dec 16, 2018
458935f
Cleanup
mjsax Mar 7, 2019
5a3613b
Github comments
mjsax Mar 7, 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
7 changes: 6 additions & 1 deletion ...s/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.internals.SessionStoreBuilder;
import org.apache.kafka.streams.state.internals.WindowStoreBuilder;
import org.apache.kafka.streams.state.internals.TimestampedWindowStoreBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -140,6 +141,8 @@ public StateStore build() {
long retentionPeriod() {
if (builder instanceof WindowStoreBuilder) {
return ((WindowStoreBuilder) builder).retentionPeriod();
} else if (builder instanceof TimestampedWindowStoreBuilder) {
return ((TimestampedWindowStoreBuilder) builder).retentionPeriod();
} else if (builder instanceof SessionStoreBuilder) {
return ((SessionStoreBuilder) builder).retentionPeriod();
} else {
Expand All @@ -160,7 +163,9 @@ private String name() {
}

private boolean isWindowStore() {
return builder instanceof WindowStoreBuilder || builder instanceof SessionStoreBuilder;
return builder instanceof WindowStoreBuilder
|| builder instanceof TimestampedWindowStoreBuilder
|| builder instanceof SessionStoreBuilder;
}

// Apparently Java strips the generics from this method because we're using the raw type for builder,
Expand Down
31 changes: 29 additions & 2 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 @@ -32,6 +32,8 @@
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.streams.state.TimestampedWindowStore;
import org.apache.kafka.streams.state.ValueAndTimestamp;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import org.apache.kafka.streams.state.internals.ThreadCache;
Expand Down Expand Up @@ -84,6 +86,8 @@ public StateStore getStateStore(final String name) {
if (global != null) {
if (global instanceof KeyValueStore) {
return new KeyValueStoreReadOnlyDecorator((KeyValueStore) global);
} else if (global instanceof TimestampedWindowStore) {
return new TimestampedWindowStoreReadOnlyDecorator((TimestampedWindowStore) global);
} else if (global instanceof WindowStore) {
return new WindowStoreReadOnlyDecorator((WindowStore) global);
} else if (global instanceof SessionStore) {
Expand All @@ -106,6 +110,8 @@ public StateStore getStateStore(final String name) {
final StateStore store = stateManager.getStore(name);
if (store instanceof KeyValueStore) {
return new KeyValueStoreReadWriteDecorator((KeyValueStore) store);
} else if (store instanceof TimestampedWindowStore) {
return new TimestampedWindowStoreReadWriteDecorator((TimestampedWindowStore) store);
} else if (store instanceof WindowStore) {
return new WindowStoreReadWriteDecorator((WindowStore) store);
} else if (store instanceof SessionStore) {
Expand Down Expand Up @@ -339,6 +345,15 @@ public KeyValueIterator<Windowed<K>, V> fetchAll(final long timeFrom,
}
}

private static class TimestampedWindowStoreReadOnlyDecorator<K, V>

@bbejeck bbejeck Jan 30, 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.

Since the new classes are static maybe they could go in their own package along with the other pre-existing decorators? Probably not on this PR but in a follow-up.

@mjsax mjsax Jan 30, 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.

I am open to add new packages -- @vvcephei was suggestion this too (for example for RocksDB classes). I would prefer to do this a follow up PRs (it's internal anyway).

extends WindowStoreReadOnlyDecorator<K, ValueAndTimestamp<V>>
implements TimestampedWindowStore<K, V> {

private TimestampedWindowStoreReadOnlyDecorator(final TimestampedWindowStore<K, V> inner) {
super(inner);
}
}

private static class SessionStoreReadOnlyDecorator<K, AGG>
extends StateStoreReadOnlyDecorator<SessionStore<K, AGG>, K, AGG>
implements SessionStore<K, AGG> {
Expand Down Expand Up @@ -520,6 +535,15 @@ public KeyValueIterator<Windowed<K>, V> fetchAll(final long timeFrom,
}
}

private static class TimestampedWindowStoreReadWriteDecorator<K, V>
extends WindowStoreReadWriteDecorator<K, ValueAndTimestamp<V>>
implements TimestampedWindowStore<K, V> {

TimestampedWindowStoreReadWriteDecorator(final TimestampedWindowStore<K, V> inner) {
super(inner);
}
}

static class SessionStoreReadWriteDecorator<K, AGG>
extends StateStoreReadWriteDecorator<SessionStore<K, AGG>, K, AGG>
implements SessionStore<K, AGG> {
Expand Down Expand Up @@ -549,12 +573,15 @@ public void remove(final Windowed<K> sessionKey) {
}

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

@Override
public AGG fetchSession(final K key, final long startTime, final long endTime) {
public AGG fetchSession(final K key,
final long startTime,
final long endTime) {
return wrapped().fetchSession(key, startTime, endTime);
}

Expand Down
47 changes: 26 additions & 21 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 @@ -64,9 +64,10 @@ private void initInternal(final InternalProcessorContext context) {
this.context = context;
final String topic = ProcessorStateManager.storeChangelogTopic(context.applicationId(), name());

bytesSerdes = new StateSerdes<>(topic,
Serdes.Bytes(),
Serdes.ByteArray());
bytesSerdes = new StateSerdes<>(
topic,
Serdes.Bytes(),
Serdes.ByteArray());
name = context.taskId() + "-" + name();
cache = this.context.getCache();

Expand Down Expand Up @@ -121,12 +122,15 @@ public boolean setFlushListener(final CacheFlushListener<byte[], byte[]> flushLi
}

@Override
public synchronized void put(final Bytes key, final byte[] value) {
public synchronized void put(final Bytes key,
final byte[] value) {
put(key, value, context.timestamp());
}

@Override
public synchronized void put(final Bytes key, final byte[] value, final long windowStartTimestamp) {
public synchronized void put(final Bytes key,
final byte[] value,
final long windowStartTimestamp) {
// since this function may not access the underlying inner store, we need to validate
// if store is open outside as well.
validateStoreOpen();
Expand All @@ -145,7 +149,8 @@ public synchronized void put(final Bytes key, final byte[] value, final long win
}

@Override
public byte[] fetch(final Bytes key, final long timestamp) {
public byte[] fetch(final Bytes key,
final long timestamp) {
validateStoreOpen();
final Bytes bytesKey = WindowKeySchema.toStoreKeyBinary(key, timestamp, 0);
final Bytes cacheKey = cacheFunction.cacheKey(bytesKey);
Expand All @@ -162,7 +167,9 @@ public byte[] fetch(final Bytes key, final long timestamp) {

@SuppressWarnings("deprecation")
@Override
public synchronized WindowStoreIterator<byte[]> fetch(final Bytes key, final long timeFrom, final long timeTo) {
public synchronized WindowStoreIterator<byte[]> fetch(final Bytes key,
final long timeFrom,
final long timeTo) {
// since this function may not access the underlying inner store, we need to validate
// if store is open outside as well.
validateStoreOpen();
Expand All @@ -175,10 +182,7 @@ public synchronized WindowStoreIterator<byte[]> fetch(final Bytes key, final lon
final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRangeFixedSize(key, timeTo));
final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, cacheKeyFrom, cacheKeyTo);

final HasNextCondition hasNextCondition = keySchema.hasNextCondition(key,
key,
timeFrom,
timeTo);
final HasNextCondition hasNextCondition = keySchema.hasNextCondition(key, key, timeFrom, timeTo);
final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(
cacheIterator, hasNextCondition, cacheFunction
);
Expand All @@ -188,23 +192,24 @@ public synchronized WindowStoreIterator<byte[]> fetch(final Bytes key, final lon

@SuppressWarnings("deprecation")
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes from, final Bytes to, final long timeFrom, final long timeTo) {
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes from,
final Bytes to,
final long timeFrom,
final long timeTo) {
// since this function may not access the underlying inner store, we need to validate
// if store is open outside as well.
validateStoreOpen();

final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator = wrapped().fetch(from, to, timeFrom, timeTo);
final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator =
wrapped().fetch(from, to, timeFrom, timeTo);
if (cache == null) {
return underlyingIterator;
}
final Bytes cacheKeyFrom = cacheFunction.cacheKey(keySchema.lowerRange(from, timeFrom));
final Bytes cacheKeyTo = cacheFunction.cacheKey(keySchema.upperRange(to, timeTo));
final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.range(name, cacheKeyFrom, cacheKeyTo);

final HasNextCondition hasNextCondition = keySchema.hasNextCondition(from,
to,
timeFrom,
timeTo);
final HasNextCondition hasNextCondition = keySchema.hasNextCondition(from, to, timeFrom, timeTo);
final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction);

return new MergedSortedCacheWindowStoreKeyValueIterator(
Expand All @@ -218,16 +223,16 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes from, final B

@SuppressWarnings("deprecation")
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom, final long timeTo) {
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom,
final long timeTo) {
validateStoreOpen();

final KeyValueIterator<Windowed<Bytes>, byte[]> underlyingIterator = wrapped().fetchAll(timeFrom, timeTo);
final ThreadCache.MemoryLRUCacheBytesIterator cacheIterator = cache.all(name);

final HasNextCondition hasNextCondition = keySchema.hasNextCondition(null, null, timeFrom, timeTo);
final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator = new FilteredCacheIterator(cacheIterator,
hasNextCondition,
cacheFunction);
final PeekingKeyValueIterator<Bytes, LRUCacheEntry> filteredCacheIterator =
new FilteredCacheIterator(cacheIterator, hasNextCondition, cacheFunction);
return new MergedSortedCacheWindowStoreKeyValueIterator(
filteredCacheIterator,
underlyingIterator,
Expand Down
1 change: 1 addition & 0 deletions .../org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedKeyValueBytesStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import static org.apache.kafka.streams.state.internals.ValueAndTimestampDeserializer.timestamp;

public class ChangeLoggingTimestampedKeyValueBytesStore extends ChangeLoggingKeyValueBytesStore {

ChangeLoggingTimestampedKeyValueBytesStore(final KeyValueStore<Bytes, byte[]> inner) {
super(inner);
}
Expand Down
41 changes: 41 additions & 0 deletions ...va/org/apache/kafka/streams/state/internals/ChangeLoggingTimestampedWindowBytesStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.kafka.streams.state.internals;

import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.state.WindowStore;

import static org.apache.kafka.streams.state.internals.ValueAndTimestampDeserializer.rawValue;
import static org.apache.kafka.streams.state.internals.ValueAndTimestampDeserializer.timestamp;

class ChangeLoggingTimestampedWindowBytesStore extends ChangeLoggingWindowBytesStore {

ChangeLoggingTimestampedWindowBytesStore(final WindowStore<Bytes, byte[]> bytesStore,
final boolean retainDuplicates) {
super(bytesStore, retainDuplicates);
}

@Override
void log(final Bytes key,
final byte[] valueAndTimestamp) {
if (valueAndTimestamp != null) {
changeLogger.logChange(key, rawValue(valueAndTimestamp), timestamp(valueAndTimestamp));
} else {
changeLogger.logChange(key, null);

@guozhangwang guozhangwang Mar 7, 2019

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.

This is not specifically related to this line of code but when I read about this, I was thinking about some correlated topic: when we call kvStore.delete and sessionStore.remove, we will pass null as ValueAndTimestamp and hence the timestamp field will also be null. Whereas in windowStore we do not have a delete api, but users can call put(k, null) or put(k, null, timestamp) in which case the timestamp field will not be null actually. Hence there's a discrepancy between the first two and the latter one, right?

@mjsax mjsax Mar 7, 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.

WindowStore#put(k, null, timestamp) is more or less the same as put(k, null) (the later actually calls the former and used context.timestamp() as third parameter) -- note that the timestamp that is passed is the windowStartTimestamp (not the timestamp that is associated with the value for the timestamped store).

In fact, we plan to deprecate and remove put(k, null), because it does not make sense semantically, to put something into a window store without specifying the window start timestamp (and context.timestamp() is not a good default value). Thus, for put(k, null, timestamp) the provided timestamp would be ignored anyway.

Also note, that for timestamped-store, the call would be put(key, valueAndTimestamp, windowStartTimestamp) and the timestamp in the value is not related to the window-start-timestamp (that will be part of the key in the store).

I agree that calling put(key, null, timestamp) to delete a window would be weird. However, in the DSL we would never delete a window anyway and don't need this. Not sure if PAPI uses would want a proper remove(key) method.

@bbejeck bbejeck Mar 7, 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.

I am just adding my 2 cents here. I understand what you are saying @mjsax, but I think if we should have a follow-up PR to update the docs to fully explain the semantic operations and differences between (kvStore, sessionStore) and the windowStore.

@mjsax mjsax Mar 7, 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.

We have many gaps in store documentation, and I agree it would be worth to do a follow up for this.

I also think I understand now why WindowStore#put(k, v) was added in the first place. For stream-steam join, we use windowed stores (with "allow duplicates" enabled) to store each individual record and thus there is not really a notion of a window (we use it more like a key-value store that allows for time-range queried plus retention time), and we use the record timestamp as "window start timestamp" (ie, we can use this shortcut method). Still think, we should remove it though.

@guozhangwang guozhangwang Mar 7, 2019

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.

Thanks for the explanation @mjsax , yes we were trying to "reuse" the window store for stream-stream windowed join, which, as an after-thought is not a very good design.

I think we do not need to make any code changes atm (as for doc changes maybe we can track those as a separate ticket / PR).

}
}
}
48 changes: 32 additions & 16 deletions ...src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingWindowBytesStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,30 +36,49 @@ class ChangeLoggingWindowBytesStore
implements WindowStore<Bytes, byte[]> {

private final boolean retainDuplicates;
private StoreChangeLogger<Bytes, byte[]> changeLogger;
private ProcessorContext context;
private int seqnum = 0;

StoreChangeLogger<Bytes, byte[]> changeLogger;

ChangeLoggingWindowBytesStore(final WindowStore<Bytes, byte[]> bytesStore,
final boolean retainDuplicates) {
super(bytesStore);
this.retainDuplicates = retainDuplicates;
}

@Override
public byte[] fetch(final Bytes key, final long timestamp) {
public void init(final ProcessorContext context,

@mjsax mjsax Mar 2, 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.

init() is just moved from below.

final StateStore root) {
this.context = context;
super.init(context, root);
final String topic = ProcessorStateManager.storeChangelogTopic(context.applicationId(), name());
changeLogger = new StoreChangeLogger<>(
name(),
context,
new StateSerdes<>(topic, Serdes.Bytes(), Serdes.ByteArray()));
}

@Override
public byte[] fetch(final Bytes key,
final long timestamp) {
return wrapped().fetch(key, timestamp);
}

@SuppressWarnings("deprecation")
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key, final long from, final long to) {
public WindowStoreIterator<byte[]> fetch(final Bytes key,
final long from,
final long to) {
return wrapped().fetch(key, from, to);
}

@SuppressWarnings("deprecation")
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes keyFrom, final Bytes keyTo, final long from, final long to) {
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes keyFrom,
final Bytes keyTo,
final long from,
final long to) {
return wrapped().fetch(keyFrom, keyTo, from, to);
}

Expand All @@ -70,7 +89,8 @@ public KeyValueIterator<Windowed<Bytes>, byte[]> all() {

@SuppressWarnings("deprecation")
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom, final long timeTo) {
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom,
final long timeTo) {
return wrapped().fetchAll(timeFrom, timeTo);
}

Expand All @@ -84,20 +104,16 @@ public void put(final Bytes key, final byte[] value) {
}

@Override
public void put(final Bytes key, final byte[] value, final long windowStartTimestamp) {
public void put(final Bytes key,
final byte[] value,
final long windowStartTimestamp) {
wrapped().put(key, value, windowStartTimestamp);
changeLogger.logChange(WindowKeySchema.toStoreKeyBinary(key, windowStartTimestamp, maybeUpdateSeqnumForDups()), value);
log(WindowKeySchema.toStoreKeyBinary(key, windowStartTimestamp, maybeUpdateSeqnumForDups()), value);
}

@Override
public void init(final ProcessorContext context, final StateStore root) {
this.context = context;
super.init(context, root);
final String topic = ProcessorStateManager.storeChangelogTopic(context.applicationId(), name());
changeLogger = new StoreChangeLogger<>(
name(),
context,
new StateSerdes<>(topic, Serdes.Bytes(), Serdes.ByteArray()));
void log(final Bytes key,
final byte[] value) {
changeLogger.logChange(key, value);
}

private int maybeUpdateSeqnumForDups() {
Expand Down
Loading