Skip to content

KAFKA-6455: Session Aggregation should use window-end-time as record timestamp #6645

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
guozhangwang merged 2 commits into from
May 12, 2019
Merged

KAFKA-6455: Session Aggregation should use window-end-time as record timestamp #6645

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
7caf7a0
KAFKA-6455: Session Aggregation should use window-end-time as record …
mjsax Apr 28, 2019
4387329
Removed unused classes after rebase
mjsax May 12, 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: 5 additions & 5 deletions ...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 @@ -78,7 +78,7 @@ public void enableSendingOldValues() {
private class KStreamSessionWindowAggregateProcessor extends AbstractProcessor<K, V> {

private SessionStore<K, Agg> store;
private TupleForwarder<Windowed<K>, Agg> tupleForwarder;
private SessionTupleForwarder<K, Agg> tupleForwarder;
private StreamsMetricsImpl metrics;
private InternalProcessorContext internalProcessorContext;
private Sensor lateRecordDropSensor;
Expand All @@ -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<>(context), sendOldValues);
tupleForwarder = new SessionTupleForwarder<>(store, context, new SessionCacheFlushListener<>(context), sendOldValues);
}

@Override
Expand All @@ -109,10 +109,10 @@ value, context().topic(), context().partition(), context().offset()
return;
}

observedStreamTime = Math.max(observedStreamTime, context().timestamp());
final long timestamp = context().timestamp();
observedStreamTime = Math.max(observedStreamTime, timestamp);
final long closeTime = observedStreamTime - windows.gracePeriodMs();

final long timestamp = context().timestamp();
final List<KeyValue<Windowed<K>, Agg>> merged = new ArrayList<>();
final SessionWindow newSessionWindow = new SessionWindow(timestamp, timestamp);
SessionWindow mergedWindow = newSessionWindow;
Expand Down Expand Up @@ -148,7 +148,7 @@ value, context().topic(), context().partition(), context().offset()
context().topic(),
context().partition(),
context().offset(),
context().timestamp(),
timestamp,
mergedWindow.start(),
mergedWindow.end(),
closeTime,
Expand Down
9 changes: 5 additions & 4 deletions ...ternals/ForwardingCacheFlushListener.java → .../internals/SessionCacheFlushListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,30 +16,31 @@
*/
package org.apache.kafka.streams.kstream.internals;

import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.To;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.apache.kafka.streams.processor.internals.ProcessorNode;
import org.apache.kafka.streams.state.internals.CacheFlushListener;

class ForwardingCacheFlushListener<K, V> implements CacheFlushListener<K, V> {
class SessionCacheFlushListener<K, V> implements CacheFlushListener<Windowed<K>, V> {
private final InternalProcessorContext context;
private final ProcessorNode myNode;

ForwardingCacheFlushListener(final ProcessorContext context) {
SessionCacheFlushListener(final ProcessorContext context) {
this.context = (InternalProcessorContext) context;
myNode = this.context.currentNode();
}

@Override
public void apply(final K key,
public void apply(final Windowed<K> key,
final V newValue,
final V oldValue,
final long timestamp) {
final ProcessorNode prev = context.currentNode();
context.setCurrentNode(myNode);
try {
context.forward(key, new Change<>(newValue, oldValue), To.all().withTimestamp(timestamp));
context.forward(key, new Change<>(newValue, oldValue), To.all().withTimestamp(key.window().end()));
} finally {
context.setCurrentNode(prev);
}
Expand Down
21 changes: 12 additions & 9 deletions ...ams/kstream/internals/TupleForwarder.java → ...ream/internals/SessionTupleForwarder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,38 +16,41 @@
*/
package org.apache.kafka.streams.kstream.internals;

import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.To;
import org.apache.kafka.streams.state.internals.CacheFlushListener;
import org.apache.kafka.streams.state.internals.WrappedStateStore;

/**
* This class is used to determine if a processor should forward values to child nodes.
* Forwarding by this class only occurs when caching is not enabled. If caching is enabled,
* forwarding occurs in the flush listener when the cached store flushes.
*
* @param <K> the type of the key
* @param <V> the type of the value
* @param <K>
* @param <V>
*/
class TupleForwarder<K, V> {
class SessionTupleForwarder<K, V> {
private final ProcessorContext context;
private final boolean sendOldValues;
private final boolean cachingEnabled;

@SuppressWarnings("unchecked")
TupleForwarder(final StateStore store,
final ProcessorContext context,
final ForwardingCacheFlushListener<K, V> flushListener,
final boolean sendOldValues) {
SessionTupleForwarder(final StateStore store,
final ProcessorContext context,
final CacheFlushListener<Windowed<K>, V> flushListener,
final boolean sendOldValues) {
this.context = context;
this.sendOldValues = sendOldValues;

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

Similar to #6667 -- we should obey sendOldValues.

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

As above

cachingEnabled = ((WrappedStateStore) store).setFlushListener(flushListener, sendOldValues);
}

public void maybeForward(final K key,
public void maybeForward(final Windowed<K> key,
final V newValue,
final V oldValue) {
if (!cachingEnabled) {
context.forward(key, new Change<>(newValue, sendOldValues ? oldValue : null));
context.forward(key, new Change<>(newValue, sendOldValues ? oldValue : null), To.all().withTimestamp(key.window().end()));
}
}
}
11 changes: 8 additions & 3 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 @@ -160,12 +160,17 @@ public <K, V> void forward(final K key,
final V value,
final To to) {
final ProcessorNode previousNode = currentNode();
final long currentTimestamp = recordContext.timestamp();
final ProcessorRecordContext previousContext = recordContext;

try {
toInternal.update(to);
if (toInternal.hasTimestamp()) {
recordContext.setTimestamp(toInternal.timestamp());
recordContext = new ProcessorRecordContext(
toInternal.timestamp(),
recordContext.offset(),
recordContext.partition(),
recordContext.topic(),
recordContext.headers());

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

@vvcephei I put this fix to make ProcessorRecordContext immutable -- however, after you comment about headers being mutable, I am wondering if this is an issue we need to address here or not. Maybe not, because it's a general issue and the users responsibility to deep-copy headers is they are modified. Just wanted to double check and point it out.

}

final String sendTo = toInternal.child();
Expand All @@ -183,7 +188,7 @@ public <K, V> void forward(final K key,
forward(child, key, value);
}
} finally {
recordContext.setTimestamp(currentTimestamp);
recordContext = previousContext;
setCurrentNode(previousNode);
}
}
Expand Down
6 changes: 1 addition & 5 deletions ...ms/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@

public class ProcessorRecordContext implements RecordContext {

private long timestamp;
private final long timestamp;
private final long offset;
private final String topic;
private final int partition;
Expand All @@ -48,10 +48,6 @@ public ProcessorRecordContext(final long timestamp,
this.headers = headers;
}

public void setTimestamp(final long timestamp) {

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

This class must be immutable -- otherwise, using context.forward(..., To.) in combination with suppress() breaks, because suppress() buffers a reference to the context and assumes it's immutable.

this.timestamp = timestamp;
}

@Override
public long offset() {
return offset;
Expand Down
4 changes: 2 additions & 2 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractStoreBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,8 +38,8 @@ public AbstractStoreBuilder(final String name,
final Serde<K> keySerde,
final Serde<V> valueSerde,
final Time time) {
Objects.requireNonNull(name, "name can't be null");
Objects.requireNonNull(time, "time can't be null");
Objects.requireNonNull(name, "name cannot be null");

@abbccdda abbccdda Apr 30, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Why we change this?

@mjsax mjsax May 1, 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 think it's improves the style. It's internal only anyway.

Objects.requireNonNull(time, "time cannot be null");
this.name = name;
this.keySerde = keySerde;
this.valueSerde = valueSerde;
Expand Down
1 change: 0 additions & 1 deletion 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 @@ -28,7 +28,6 @@
import org.apache.kafka.streams.processor.internals.RecordQueue;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.SessionStore;

import java.util.Objects;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down
19 changes: 10 additions & 9 deletions streams/src/main/java/org/apache/kafka/streams/state/internals/SessionStoreBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,8 @@
import org.apache.kafka.streams.state.SessionBytesStoreSupplier;
import org.apache.kafka.streams.state.SessionStore;

import java.util.Objects;


public class SessionStoreBuilder<K, V> extends AbstractStoreBuilder<K, V, SessionStore<K, V>> {

Expand All @@ -31,26 +33,25 @@ public SessionStoreBuilder(final SessionBytesStoreSupplier storeSupplier,
final Serde<K> keySerde,
final Serde<V> valueSerde,
final Time time) {
super(storeSupplier.name(), keySerde, valueSerde, time);
super(Objects.requireNonNull(storeSupplier, "supplier cannot be null").name(), keySerde, valueSerde, time);
this.storeSupplier = storeSupplier;
}

@Override
public SessionStore<K, V> build() {
return new MeteredSessionStore<>(maybeWrapCaching(maybeWrapLogging(storeSupplier.get())),
storeSupplier.metricsScope(),
keySerde,
valueSerde,
time);
return new MeteredSessionStore<>(
maybeWrapCaching(maybeWrapLogging(storeSupplier.get())),
storeSupplier.metricsScope(),
keySerde,
valueSerde,
time);
}

private SessionStore<Bytes, byte[]> maybeWrapCaching(final SessionStore<Bytes, byte[]> inner) {
if (!enableCaching) {
return inner;
}
return new CachingSessionStore(
inner,
storeSupplier.segmentIntervalMs());
return new CachingSessionStore(inner, storeSupplier.segmentIntervalMs());
}

private SessionStore<Bytes, byte[]> maybeWrapLogging(final SessionStore<Bytes, byte[]> inner) {
Expand Down
Loading