You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mj...@apache.org on 2018/10/02 06:10:42 UTC
[kafka] branch trunk updated: KAFKA-7223: In-Memory Suppression
Buffering (#5693)
This is an automated email from the ASF dual-hosted git repository.
mjsax pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 5ba9cad KAFKA-7223: In-Memory Suppression Buffering (#5693)
5ba9cad is described below
commit 5ba9cade7b066cc26842aeaac5662a57c502ffcb
Author: John Roesler <vv...@users.noreply.github.com>
AuthorDate: Tue Oct 2 01:10:34 2018 -0500
KAFKA-7223: In-Memory Suppression Buffering (#5693)
Reviewer: Bill Bejeck <bi...@confluent.io>, Guozhang Wang <gu...@confluent.io>, Matthias J. Sax <ma...@confluent.io>
---
.../apache/kafka/streams/kstream/Suppressed.java | 4 +-
.../streams/kstream/internals/FullChangeSerde.java | 23 +-
.../kstream/internals/FullTimeWindowedSerde.java | 32 +++
.../streams/kstream/internals/KTableImpl.java | 17 +-
.../kstream/internals/TimeWindowedKStreamImpl.java | 7 +-
...erConfigImpl.java => BufferConfigInternal.java} | 8 +-
.../internals/suppress/EagerBufferConfigImpl.java | 20 +-
.../suppress/FinalResultsSuppressionBuilder.java | 7 +-
.../InMemoryTimeOrderedKeyValueBuffer.java | 116 ++++++++++
.../suppress/KTableSuppressProcessor.java | 111 +++++++---
.../internals/suppress/StrictBufferConfigImpl.java | 22 +-
...SuppressedImpl.java => SuppressedInternal.java} | 36 +--
.../internals/suppress/TimeDefinitions.java | 79 +++++++
...ResultsSuppressionBuilder.java => TimeKey.java} | 41 ++--
.../suppress/TimeOrderedKeyValueBuffer.java | 36 +++
.../internals/ProcessorRecordContext.java | 18 ++
.../state/internals/CachingKeyValueStore.java | 2 +-
.../state/internals/CachingSessionStore.java | 2 +-
.../state/internals/CachingWindowStore.java | 2 +-
.../streams/state/internals/ContextualRecord.java | 58 +++++
.../streams/state/internals/LRUCacheEntry.java | 42 ++--
.../kafka/streams/state/internals/ThreadCache.java | 7 +-
.../integration/SuppressionIntegrationTest.java | 96 +++++++-
.../kafka/streams/kstream/SuppressedTest.java | 12 +-
.../kstream/internals/SuppressScenarioTest.java | 48 +---
.../suppress/KTableSuppressProcessorTest.java | 245 ++++++++++++++++-----
.../streams/state/internals/NamedCacheTest.java | 2 +-
27 files changed, 841 insertions(+), 252 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Suppressed.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Suppressed.java
index 49fe96b..6db7a70 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Suppressed.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/Suppressed.java
@@ -19,7 +19,7 @@ package org.apache.kafka.streams.kstream;
import org.apache.kafka.streams.kstream.internals.suppress.EagerBufferConfigImpl;
import org.apache.kafka.streams.kstream.internals.suppress.FinalResultsSuppressionBuilder;
import org.apache.kafka.streams.kstream.internals.suppress.StrictBufferConfigImpl;
-import org.apache.kafka.streams.kstream.internals.suppress.SuppressedImpl;
+import org.apache.kafka.streams.kstream.internals.suppress.SuppressedInternal;
import java.time.Duration;
@@ -155,6 +155,6 @@ public interface Suppressed<K> {
* @return a suppression configuration
*/
static <K> Suppressed<K> untilTimeLimit(final Duration timeToWaitForMoreEvents, final BufferConfig bufferConfig) {
- return new SuppressedImpl<>(timeToWaitForMoreEvents, bufferConfig, null, false);
+ return new SuppressedInternal<>(timeToWaitForMoreEvents, bufferConfig, null, false);
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java
index 8a2e619..9bb8373 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullChangeSerde.java
@@ -16,8 +16,6 @@
*/
package org.apache.kafka.streams.kstream.internals;
-import org.apache.kafka.common.serialization.ByteBufferDeserializer;
-import org.apache.kafka.common.serialization.ByteBufferSerializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serializer;
@@ -30,6 +28,17 @@ import static java.util.Objects.requireNonNull;
public class FullChangeSerde<T> implements Serde<Change<T>> {
private final Serde<T> inner;
+ @SuppressWarnings("unchecked")
+ public static <T> FullChangeSerde<T> castOrWrap(final Serde<?> serde) {
+ if (serde == null) {
+ return null;
+ } else if (serde instanceof FullChangeSerde) {
+ return (FullChangeSerde<T>) serde;
+ } else {
+ return new FullChangeSerde<T>((Serde<T>) serde);
+ }
+ }
+
public FullChangeSerde(final Serde<T> inner) {
this.inner = requireNonNull(inner);
}
@@ -47,7 +56,6 @@ public class FullChangeSerde<T> implements Serde<Change<T>> {
@Override
public Serializer<Change<T>> serializer() {
final Serializer<T> innerSerializer = inner.serializer();
- final ByteBufferSerializer byteBufferSerializer = new ByteBufferSerializer();
return new Serializer<Change<T>>() {
@Override
@@ -65,8 +73,8 @@ public class FullChangeSerde<T> implements Serde<Change<T>> {
final byte[] newBytes = data.newValue == null ? null : innerSerializer.serialize(topic, data.newValue);
final int newSize = newBytes == null ? -1 : newBytes.length;
- final ByteBuffer buffer = ByteBuffer.allocate(
- 4 + (oldSize == -1 ? 0 : oldSize) + 4 + (newSize == -1 ? 0 : newSize)
+ final ByteBuffer buffer = ByteBuffer.wrap(
+ new byte[4 + (oldSize == -1 ? 0 : oldSize) + 4 + (newSize == -1 ? 0 : newSize)]
);
buffer.putInt(oldSize);
if (oldBytes != null) {
@@ -76,7 +84,7 @@ public class FullChangeSerde<T> implements Serde<Change<T>> {
if (newBytes != null) {
buffer.put(newBytes);
}
- return byteBufferSerializer.serialize(null, buffer);
+ return buffer.array();
}
@Override
@@ -89,7 +97,6 @@ public class FullChangeSerde<T> implements Serde<Change<T>> {
@Override
public Deserializer<Change<T>> deserializer() {
final Deserializer<T> innerDeserializer = inner.deserializer();
- final ByteBufferDeserializer byteBufferDeserializer = new ByteBufferDeserializer();
return new Deserializer<Change<T>>() {
@Override
public void configure(final Map<String, ?> configs, final boolean isKey) {
@@ -101,7 +108,7 @@ public class FullChangeSerde<T> implements Serde<Change<T>> {
if (data == null) {
return null;
}
- final ByteBuffer buffer = byteBufferDeserializer.deserialize(null, data);
+ final ByteBuffer buffer = ByteBuffer.wrap(data);
final int oldSize = buffer.getInt();
final byte[] oldBytes = oldSize == -1 ? null : new byte[oldSize];
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullTimeWindowedSerde.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullTimeWindowedSerde.java
new file mode 100644
index 0000000..a69002f
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/FullTimeWindowedSerde.java
@@ -0,0 +1,32 @@
+/*
+ * 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.kstream.internals;
+
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
+import org.apache.kafka.streams.kstream.TimeWindowedSerializer;
+import org.apache.kafka.streams.kstream.Windowed;
+
+class FullTimeWindowedSerde<T> extends Serdes.WrapperSerde<Windowed<T>> {
+ FullTimeWindowedSerde(final Serde<T> inner, final long windowSize) {
+ super(
+ new TimeWindowedSerializer<>(inner.serializer()),
+ new TimeWindowedDeserializer<>(inner.deserializer(), windowSize)
+ );
+ }
+}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
index 53e7a4b..3ce962b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/KTableImpl.java
@@ -40,7 +40,7 @@ import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
import org.apache.kafka.streams.kstream.internals.graph.TableProcessorNode;
import org.apache.kafka.streams.kstream.internals.suppress.FinalResultsSuppressionBuilder;
import org.apache.kafka.streams.kstream.internals.suppress.KTableSuppressProcessor;
-import org.apache.kafka.streams.kstream.internals.suppress.SuppressedImpl;
+import org.apache.kafka.streams.kstream.internals.suppress.SuppressedInternal;
import org.apache.kafka.streams.processor.ProcessorSupplier;
import org.apache.kafka.streams.state.KeyValueStore;
@@ -357,12 +357,11 @@ public class KTableImpl<K, S, V> extends AbstractStream<K, V> implements KTable<
public KTable<K, V> suppress(final Suppressed<K> suppressed) {
final String name = builder.newProcessorName(SUPPRESS_NAME);
- // TODO: follow-up pr to forward the k/v serdes
final ProcessorSupplier<K, Change<V>> suppressionSupplier =
() -> new KTableSuppressProcessor<>(
buildSuppress(suppressed),
- null,
- null
+ keySerde,
+ valSerde == null ? null : new FullChangeSerde<>(valSerde)
);
final ProcessorParameters<K, Change<V>> processorParameters = new ProcessorParameters<>(
@@ -388,18 +387,18 @@ public class KTableImpl<K, S, V> extends AbstractStream<K, V> implements KTable<
}
@SuppressWarnings("unchecked")
- private SuppressedImpl<K> buildSuppress(final Suppressed<K> suppress) {
+ private SuppressedInternal<K> buildSuppress(final Suppressed<K> suppress) {
if (suppress instanceof FinalResultsSuppressionBuilder) {
final long grace = findAndVerifyWindowGrace(streamsGraphNode);
final FinalResultsSuppressionBuilder<?> builder = (FinalResultsSuppressionBuilder) suppress;
- final SuppressedImpl<? extends Windowed> finalResultsSuppression =
+ final SuppressedInternal<? extends Windowed> finalResultsSuppression =
builder.buildFinalResultsSuppression(Duration.ofMillis(grace));
- return (SuppressedImpl<K>) finalResultsSuppression;
- } else if (suppress instanceof SuppressedImpl) {
- return (SuppressedImpl<K>) suppress;
+ return (SuppressedInternal<K>) finalResultsSuppression;
+ } else if (suppress instanceof SuppressedInternal) {
+ return (SuppressedInternal<K>) suppress;
} else {
throw new IllegalArgumentException("Custom subclasses of Suppressed are not allowed.");
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java
index 8519671..fcb9c02 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/TimeWindowedKStreamImpl.java
@@ -27,7 +27,6 @@ import org.apache.kafka.streams.kstream.Reducer;
import org.apache.kafka.streams.kstream.TimeWindowedKStream;
import org.apache.kafka.streams.kstream.Window;
import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.kstream.WindowedSerdes;
import org.apache.kafka.streams.kstream.Windows;
import org.apache.kafka.streams.kstream.internals.graph.StreamsGraphNode;
import org.apache.kafka.streams.state.StoreBuilder;
@@ -93,7 +92,7 @@ public class TimeWindowedKStreamImpl<K, V, W extends Window> extends AbstractStr
materialize(materializedInternal),
new KStreamWindowAggregate<>(windows, materializedInternal.storeName(), aggregateBuilder.countInitializer, aggregateBuilder.countAggregator),
materializedInternal.isQueryable(),
- materializedInternal.keySerde() != null ? new WindowedSerdes.TimeWindowedSerde<>(materializedInternal.keySerde()) : null,
+ materializedInternal.keySerde() != null ? new FullTimeWindowedSerde<>(materializedInternal.keySerde(), windows.size()) : null,
materializedInternal.valueSerde());
}
@@ -120,7 +119,7 @@ public class TimeWindowedKStreamImpl<K, V, W extends Window> extends AbstractStr
materialize(materializedInternal),
new KStreamWindowAggregate<>(windows, materializedInternal.storeName(), initializer, aggregator),
materializedInternal.isQueryable(),
- materializedInternal.keySerde() != null ? new WindowedSerdes.TimeWindowedSerde<>(materializedInternal.keySerde()) : null,
+ materializedInternal.keySerde() != null ? new FullTimeWindowedSerde<>(materializedInternal.keySerde(), windows.size()) : null,
materializedInternal.valueSerde());
}
@@ -149,7 +148,7 @@ public class TimeWindowedKStreamImpl<K, V, W extends Window> extends AbstractStr
materialize(materializedInternal),
new KStreamWindowReduce<>(windows, materializedInternal.storeName(), reducer),
materializedInternal.isQueryable(),
- materializedInternal.keySerde() != null ? new WindowedSerdes.TimeWindowedSerde<>(materializedInternal.keySerde()) : null,
+ materializedInternal.keySerde() != null ? new FullTimeWindowedSerde<>(materializedInternal.keySerde(), windows.size()) : null,
materializedInternal.valueSerde());
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/BufferConfigImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/BufferConfigInternal.java
similarity index 84%
rename from streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/BufferConfigImpl.java
rename to streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/BufferConfigInternal.java
index e731dc6..67d3783 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/BufferConfigImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/BufferConfigInternal.java
@@ -20,8 +20,8 @@ import org.apache.kafka.streams.kstream.Suppressed;
import static org.apache.kafka.streams.kstream.internals.suppress.BufferFullStrategy.SHUT_DOWN;
-abstract class BufferConfigImpl<BC extends Suppressed.BufferConfig<BC>> implements Suppressed.BufferConfig<BC> {
- public abstract long maxKeys();
+abstract class BufferConfigInternal<BC extends Suppressed.BufferConfig<BC>> implements Suppressed.BufferConfig<BC> {
+ public abstract long maxRecords();
public abstract long maxBytes();
@@ -39,12 +39,12 @@ abstract class BufferConfigImpl<BC extends Suppressed.BufferConfig<BC>> implemen
@Override
public Suppressed.StrictBufferConfig shutDownWhenFull() {
- return new StrictBufferConfigImpl(maxKeys(), maxBytes(), SHUT_DOWN);
+ return new StrictBufferConfigImpl(maxRecords(), maxBytes(), SHUT_DOWN);
}
@Override
public Suppressed.BufferConfig emitEarlyWhenFull() {
- return new EagerBufferConfigImpl(maxKeys(), maxBytes());
+ return new EagerBufferConfigImpl(maxRecords(), maxBytes());
}
@Override
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java
index 0c2c883..161f934 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/EagerBufferConfigImpl.java
@@ -20,13 +20,13 @@ import org.apache.kafka.streams.kstream.Suppressed;
import java.util.Objects;
-public class EagerBufferConfigImpl extends BufferConfigImpl {
+public class EagerBufferConfigImpl extends BufferConfigInternal {
- private final long maxKeys;
+ private final long maxRecords;
private final long maxBytes;
- public EagerBufferConfigImpl(final long maxKeys, final long maxBytes) {
- this.maxKeys = maxKeys;
+ public EagerBufferConfigImpl(final long maxRecords, final long maxBytes) {
+ this.maxRecords = maxRecords;
this.maxBytes = maxBytes;
}
@@ -37,12 +37,12 @@ public class EagerBufferConfigImpl extends BufferConfigImpl {
@Override
public Suppressed.BufferConfig withMaxBytes(final long byteLimit) {
- return new EagerBufferConfigImpl(maxKeys, byteLimit);
+ return new EagerBufferConfigImpl(maxRecords, byteLimit);
}
@Override
- public long maxKeys() {
- return maxKeys;
+ public long maxRecords() {
+ return maxRecords;
}
@Override
@@ -60,17 +60,17 @@ public class EagerBufferConfigImpl extends BufferConfigImpl {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
final EagerBufferConfigImpl that = (EagerBufferConfigImpl) o;
- return maxKeys == that.maxKeys &&
+ return maxRecords == that.maxRecords &&
maxBytes == that.maxBytes;
}
@Override
public int hashCode() {
- return Objects.hash(maxKeys, maxBytes);
+ return Objects.hash(maxRecords, maxBytes);
}
@Override
public String toString() {
- return "EagerBufferConfigImpl{maxKeys=" + maxKeys + ", maxBytes=" + maxBytes + '}';
+ return "EagerBufferConfigImpl{maxKeys=" + maxRecords + ", maxBytes=" + maxBytes + '}';
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/FinalResultsSuppressionBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/FinalResultsSuppressionBuilder.java
index db09307..523ae06 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/FinalResultsSuppressionBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/FinalResultsSuppressionBuilder.java
@@ -18,7 +18,6 @@ package org.apache.kafka.streams.kstream.internals.suppress;
import org.apache.kafka.streams.kstream.Suppressed;
import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.processor.ProcessorContext;
import java.time.Duration;
import java.util.Objects;
@@ -30,11 +29,11 @@ public class FinalResultsSuppressionBuilder<K extends Windowed> implements Suppr
this.bufferConfig = bufferConfig;
}
- public SuppressedImpl<K> buildFinalResultsSuppression(final Duration gracePeriod) {
- return new SuppressedImpl<>(
+ public SuppressedInternal<K> buildFinalResultsSuppression(final Duration gracePeriod) {
+ return new SuppressedInternal<>(
gracePeriod,
bufferConfig,
- (ProcessorContext context, K key) -> key.window().end(),
+ TimeDefinitions.WindowEndTimeDefinition.instance(),
true
);
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/InMemoryTimeOrderedKeyValueBuffer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/InMemoryTimeOrderedKeyValueBuffer.java
new file mode 100644
index 0000000..677a662
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/InMemoryTimeOrderedKeyValueBuffer.java
@@ -0,0 +1,116 @@
+/*
+ * 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.kstream.internals.suppress;
+
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.internals.ContextualRecord;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.TreeMap;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+class InMemoryTimeOrderedKeyValueBuffer implements TimeOrderedKeyValueBuffer {
+ private final Map<Bytes, TimeKey> index = new HashMap<>();
+ private final TreeMap<TimeKey, ContextualRecord> sortedMap = new TreeMap<>();
+ private long memBufferSize = 0L;
+ private long minTimestamp = Long.MAX_VALUE;
+
+ @Override
+ public void evictWhile(final Supplier<Boolean> predicate,
+ final Consumer<KeyValue<Bytes, ContextualRecord>> callback) {
+ final Iterator<Map.Entry<TimeKey, ContextualRecord>> delegate = sortedMap.entrySet().iterator();
+
+ if (predicate.get()) {
+ Map.Entry<TimeKey, ContextualRecord> next = null;
+ if (delegate.hasNext()) {
+ next = delegate.next();
+ }
+
+ // predicate being true means we read one record, call the callback, and then remove it
+ while (next != null && predicate.get()) {
+ callback.accept(new KeyValue<>(next.getKey().key(), next.getValue()));
+
+ delegate.remove();
+ index.remove(next.getKey().key());
+
+ memBufferSize = memBufferSize - computeRecordSize(next.getKey().key(), next.getValue());
+
+ // peek at the next record so we can update the minTimestamp
+ if (delegate.hasNext()) {
+ next = delegate.next();
+ minTimestamp = next == null ? Long.MAX_VALUE : next.getKey().time();
+ } else {
+ next = null;
+ minTimestamp = Long.MAX_VALUE;
+ }
+ }
+ }
+ }
+
+ @Override
+ public void put(final long time,
+ final Bytes key,
+ final ContextualRecord value) {
+ // non-resetting semantics:
+ // if there was a previous version of the same record,
+ // then insert the new record in the same place in the priority queue
+
+ final TimeKey previousKey = index.get(key);
+ if (previousKey == null) {
+ final TimeKey nextKey = new TimeKey(time, key);
+ index.put(key, nextKey);
+ sortedMap.put(nextKey, value);
+ minTimestamp = Math.min(minTimestamp, time);
+ memBufferSize = memBufferSize + computeRecordSize(key, value);
+ } else {
+ final ContextualRecord removedValue = sortedMap.put(previousKey, value);
+ memBufferSize =
+ memBufferSize
+ + computeRecordSize(key, value)
+ - (removedValue == null ? 0 : computeRecordSize(key, removedValue));
+ }
+ }
+
+ @Override
+ public int numRecords() {
+ return index.size();
+ }
+
+ @Override
+ public long bufferSize() {
+ return memBufferSize;
+ }
+
+ @Override
+ public long minTimestamp() {
+ return minTimestamp;
+ }
+
+ private long computeRecordSize(final Bytes key, final ContextualRecord value) {
+ long size = 0L;
+ size += 8; // buffer time
+ size += key.get().length;
+ if (value != null) {
+ size += value.sizeBytes();
+ }
+ return size;
+ }
+}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java
index 6f0021f..57e5066 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessor.java
@@ -17,70 +17,117 @@
package org.apache.kafka.streams.kstream.internals.suppress;
import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.internals.Change;
+import org.apache.kafka.streams.kstream.internals.FullChangeSerde;
+import org.apache.kafka.streams.kstream.internals.suppress.TimeDefinitions.TimeDefinition;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
-
-import java.time.Duration;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.state.internals.ContextualRecord;
import static java.util.Objects.requireNonNull;
public class KTableSuppressProcessor<K, V> implements Processor<K, Change<V>> {
- private final SuppressedImpl<K> suppress;
+ private final long maxRecords;
+ private final long maxBytes;
+ private final long suppressDurationMillis;
+ private final TimeOrderedKeyValueBuffer buffer;
+ private final TimeDefinition<K> bufferTimeDefinition;
+ private final BufferFullStrategy bufferFullStrategy;
+ private final boolean shouldSuppressTombstones;
private InternalProcessorContext internalProcessorContext;
- private final Serde<K> keySerde;
- private final Serde<Change<V>> valueSerde;
+ private Serde<K> keySerde;
+ private Serde<Change<V>> valueSerde;
- public KTableSuppressProcessor(final SuppressedImpl<K> suppress,
+ public KTableSuppressProcessor(final SuppressedInternal<K> suppress,
final Serde<K> keySerde,
- final Serde<Change<V>> valueSerde) {
- this.suppress = requireNonNull(suppress);
+ final FullChangeSerde<V> valueSerde) {
+ requireNonNull(suppress);
this.keySerde = keySerde;
this.valueSerde = valueSerde;
+ maxRecords = suppress.getBufferConfig().maxRecords();
+ maxBytes = suppress.getBufferConfig().maxBytes();
+ suppressDurationMillis = suppress.getTimeToWaitForMoreEvents().toMillis();
+ buffer = new InMemoryTimeOrderedKeyValueBuffer();
+ bufferTimeDefinition = suppress.getTimeDefinition();
+ bufferFullStrategy = suppress.getBufferConfig().bufferFullStrategy();
+ shouldSuppressTombstones = suppress.shouldSuppressTombstones();
}
+ @SuppressWarnings("unchecked")
@Override
public void init(final ProcessorContext context) {
internalProcessorContext = (InternalProcessorContext) context;
+ this.keySerde = keySerde == null ? (Serde<K>) context.keySerde() : keySerde;
+ this.valueSerde = valueSerde == null ? FullChangeSerde.castOrWrap(context.valueSerde()) : valueSerde;
}
@Override
public void process(final K key, final Change<V> value) {
- if (suppress.getTimeToWaitForMoreEvents() == Duration.ZERO && definedRecordTime(key) <= internalProcessorContext.streamTime()) {
- if (shouldForward(value)) {
- internalProcessorContext.forward(key, value);
- } // else skip
- } else {
- throw new NotImplementedException();
- }
+ buffer(key, value);
+ enforceConstraints();
}
- private boolean shouldForward(final Change<V> value) {
- return !(value.newValue == null && suppress.suppressTombstones());
- }
+ private void buffer(final K key, final Change<V> value) {
+ final long bufferTime = bufferTimeDefinition.time(internalProcessorContext, key);
+ final ProcessorRecordContext recordContext = internalProcessorContext.recordContext();
- private long definedRecordTime(final K key) {
- return suppress.getTimeDefinition().time(internalProcessorContext, key);
+ final Bytes serializedKey = Bytes.wrap(keySerde.serializer().serialize(null, key));
+ final byte[] serializedValue = valueSerde.serializer().serialize(null, value);
+
+ buffer.put(bufferTime, serializedKey, new ContextualRecord(serializedValue, recordContext));
}
- @Override
- public void close() {
+ private void enforceConstraints() {
+ final long streamTime = internalProcessorContext.streamTime();
+ final long expiryTime = streamTime - suppressDurationMillis;
+
+ buffer.evictWhile(() -> buffer.minTimestamp() <= expiryTime, this::emit);
+
+ if (overCapacity()) {
+ switch (bufferFullStrategy) {
+ case EMIT:
+ buffer.evictWhile(this::overCapacity, this::emit);
+ return;
+ case SHUT_DOWN:
+ throw new StreamsException(String.format(
+ "%s buffer exceeded its max capacity. Currently [%d/%d] records and [%d/%d] bytes.",
+ internalProcessorContext.currentNode().name(),
+ buffer.numRecords(), maxRecords,
+ buffer.bufferSize(), maxBytes
+ ));
+ }
+ }
}
- @Override
- public String toString() {
- return "KTableSuppressProcessor{" +
- "suppress=" + suppress +
- ", keySerde=" + keySerde +
- ", valueSerde=" + valueSerde +
- '}';
+ private boolean overCapacity() {
+ return buffer.numRecords() > maxRecords || buffer.bufferSize() > maxBytes;
}
- public static class NotImplementedException extends RuntimeException {
- NotImplementedException() {
- super();
+ private void emit(final KeyValue<Bytes, ContextualRecord> toEmit) {
+ final Change<V> value = valueSerde.deserializer().deserialize(null, toEmit.value.value());
+ if (shouldForward(value)) {
+ final ProcessorRecordContext prevRecordContext = internalProcessorContext.recordContext();
+ internalProcessorContext.setRecordContext(toEmit.value.recordContext());
+ try {
+ final K key = keySerde.deserializer().deserialize(null, toEmit.key.get());
+ internalProcessorContext.forward(key, value);
+ } finally {
+ internalProcessorContext.setRecordContext(prevRecordContext);
+ }
}
}
+
+ private boolean shouldForward(final Change<V> value) {
+ return !(value.newValue == null && shouldSuppressTombstones);
+ }
+
+ @Override
+ public void close() {
+ }
}
\ No newline at end of file
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/StrictBufferConfigImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/StrictBufferConfigImpl.java
index 0634a74..ef754ec 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/StrictBufferConfigImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/StrictBufferConfigImpl.java
@@ -22,22 +22,22 @@ import java.util.Objects;
import static org.apache.kafka.streams.kstream.internals.suppress.BufferFullStrategy.SHUT_DOWN;
-public class StrictBufferConfigImpl extends BufferConfigImpl<Suppressed.StrictBufferConfig> implements Suppressed.StrictBufferConfig {
+public class StrictBufferConfigImpl extends BufferConfigInternal<Suppressed.StrictBufferConfig> implements Suppressed.StrictBufferConfig {
- private final long maxKeys;
+ private final long maxRecords;
private final long maxBytes;
private final BufferFullStrategy bufferFullStrategy;
- public StrictBufferConfigImpl(final long maxKeys,
+ public StrictBufferConfigImpl(final long maxRecords,
final long maxBytes,
final BufferFullStrategy bufferFullStrategy) {
- this.maxKeys = maxKeys;
+ this.maxRecords = maxRecords;
this.maxBytes = maxBytes;
this.bufferFullStrategy = bufferFullStrategy;
}
public StrictBufferConfigImpl() {
- this.maxKeys = Long.MAX_VALUE;
+ this.maxRecords = Long.MAX_VALUE;
this.maxBytes = Long.MAX_VALUE;
this.bufferFullStrategy = SHUT_DOWN;
}
@@ -49,12 +49,12 @@ public class StrictBufferConfigImpl extends BufferConfigImpl<Suppressed.StrictBu
@Override
public Suppressed.StrictBufferConfig withMaxBytes(final long byteLimit) {
- return new StrictBufferConfigImpl(maxKeys, byteLimit, bufferFullStrategy);
+ return new StrictBufferConfigImpl(maxRecords, byteLimit, bufferFullStrategy);
}
@Override
- public long maxKeys() {
- return maxKeys;
+ public long maxRecords() {
+ return maxRecords;
}
@Override
@@ -72,19 +72,19 @@ public class StrictBufferConfigImpl extends BufferConfigImpl<Suppressed.StrictBu
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
final StrictBufferConfigImpl that = (StrictBufferConfigImpl) o;
- return maxKeys == that.maxKeys &&
+ return maxRecords == that.maxRecords &&
maxBytes == that.maxBytes &&
bufferFullStrategy == that.bufferFullStrategy;
}
@Override
public int hashCode() {
- return Objects.hash(maxKeys, maxBytes, bufferFullStrategy);
+ return Objects.hash(maxRecords, maxBytes, bufferFullStrategy);
}
@Override
public String toString() {
- return "StrictBufferConfigImpl{maxKeys=" + maxKeys +
+ return "StrictBufferConfigImpl{maxKeys=" + maxRecords +
", maxBytes=" + maxBytes +
", bufferFullStrategy=" + bufferFullStrategy + '}';
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedImpl.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java
similarity index 71%
rename from streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedImpl.java
rename to streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java
index a3bf2db..99245da 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedImpl.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/SuppressedInternal.java
@@ -17,32 +17,32 @@
package org.apache.kafka.streams.kstream.internals.suppress;
import org.apache.kafka.streams.kstream.Suppressed;
-import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.kstream.internals.suppress.TimeDefinitions.TimeDefinition;
import java.time.Duration;
import java.util.Objects;
-public class SuppressedImpl<K> implements Suppressed<K> {
+public class SuppressedInternal<K> implements Suppressed<K> {
private static final Duration DEFAULT_SUPPRESSION_TIME = Duration.ofMillis(Long.MAX_VALUE);
- private static final StrictBufferConfig DEFAULT_BUFFER_CONFIG = BufferConfig.unbounded();
+ private static final StrictBufferConfigImpl DEFAULT_BUFFER_CONFIG = (StrictBufferConfigImpl) BufferConfig.unbounded();
- private final BufferConfig bufferConfig;
+ private final BufferConfigInternal bufferConfig;
private final Duration timeToWaitForMoreEvents;
private final TimeDefinition<K> timeDefinition;
private final boolean suppressTombstones;
- public SuppressedImpl(final Duration suppressionTime,
- final BufferConfig bufferConfig,
- final TimeDefinition<K> timeDefinition,
- final boolean suppressTombstones) {
+ public SuppressedInternal(final Duration suppressionTime,
+ final BufferConfig bufferConfig,
+ final TimeDefinition<K> timeDefinition,
+ final boolean suppressTombstones) {
this.timeToWaitForMoreEvents = suppressionTime == null ? DEFAULT_SUPPRESSION_TIME : suppressionTime;
- this.timeDefinition = timeDefinition == null ? (context, anyKey) -> context.timestamp() : timeDefinition;
- this.bufferConfig = bufferConfig == null ? DEFAULT_BUFFER_CONFIG : bufferConfig;
+ this.timeDefinition = timeDefinition == null ? TimeDefinitions.RecordTimeDefintion.instance() : timeDefinition;
+ this.bufferConfig = bufferConfig == null ? DEFAULT_BUFFER_CONFIG : (BufferConfigInternal) bufferConfig;
this.suppressTombstones = suppressTombstones;
}
- interface TimeDefinition<K> {
- long time(final ProcessorContext context, final K key);
+ BufferConfigInternal getBufferConfig() {
+ return bufferConfig;
}
TimeDefinition<K> getTimeDefinition() {
@@ -53,11 +53,15 @@ public class SuppressedImpl<K> implements Suppressed<K> {
return timeToWaitForMoreEvents == null ? Duration.ZERO : timeToWaitForMoreEvents;
}
+ boolean shouldSuppressTombstones() {
+ return suppressTombstones;
+ }
+
@Override
public boolean equals(final Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- final SuppressedImpl<?> that = (SuppressedImpl<?>) o;
+ final SuppressedInternal<?> that = (SuppressedInternal<?>) o;
return Objects.equals(bufferConfig, that.bufferConfig) &&
Objects.equals(getTimeToWaitForMoreEvents(), that.getTimeToWaitForMoreEvents()) &&
Objects.equals(getTimeDefinition(), that.getTimeDefinition());
@@ -70,14 +74,10 @@ public class SuppressedImpl<K> implements Suppressed<K> {
@Override
public String toString() {
- return "SuppressedImpl{" +
+ return "SuppressedInternal{" +
", bufferConfig=" + bufferConfig +
", timeToWaitForMoreEvents=" + timeToWaitForMoreEvents +
", timeDefinition=" + timeDefinition +
'}';
}
-
- boolean suppressTombstones() {
- return suppressTombstones;
- }
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeDefinitions.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeDefinitions.java
new file mode 100644
index 0000000..b37bcf6
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeDefinitions.java
@@ -0,0 +1,79 @@
+/*
+ * 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.kstream.internals.suppress;
+
+import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.processor.ProcessorContext;
+
+final class TimeDefinitions {
+ private TimeDefinitions() {}
+
+ enum TimeDefinitionType {
+ RECORD_TIME, WINDOW_END_TIME;
+ }
+
+ /**
+ * This interface should never be instantiated outside of this class.
+ */
+ interface TimeDefinition<K> {
+ long time(final ProcessorContext context, final K key);
+
+ TimeDefinitionType type();
+ }
+
+ public static class RecordTimeDefintion<K> implements TimeDefinition<K> {
+ private static final RecordTimeDefintion INSTANCE = new RecordTimeDefintion();
+
+ private RecordTimeDefintion() {}
+
+ @SuppressWarnings("unchecked")
+ public static <K> RecordTimeDefintion<K> instance() {
+ return RecordTimeDefintion.INSTANCE;
+ }
+
+ @Override
+ public long time(final ProcessorContext context, final K key) {
+ return context.timestamp();
+ }
+
+ @Override
+ public TimeDefinitionType type() {
+ return TimeDefinitionType.RECORD_TIME;
+ }
+ }
+
+ public static class WindowEndTimeDefinition<K extends Windowed> implements TimeDefinition<K> {
+ private static final WindowEndTimeDefinition INSTANCE = new WindowEndTimeDefinition();
+
+ private WindowEndTimeDefinition() {}
+
+ @SuppressWarnings("unchecked")
+ public static <K extends Windowed> WindowEndTimeDefinition<K> instance() {
+ return WindowEndTimeDefinition.INSTANCE;
+ }
+
+ @Override
+ public long time(final ProcessorContext context, final K key) {
+ return key.window().end();
+ }
+
+ @Override
+ public TimeDefinitionType type() {
+ return TimeDefinitionType.WINDOW_END_TIME;
+ }
+ }
+}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/FinalResultsSuppressionBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeKey.java
similarity index 51%
copy from streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/FinalResultsSuppressionBuilder.java
copy to streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeKey.java
index db09307..d3ad350 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/FinalResultsSuppressionBuilder.java
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeKey.java
@@ -16,44 +16,45 @@
*/
package org.apache.kafka.streams.kstream.internals.suppress;
-import org.apache.kafka.streams.kstream.Suppressed;
-import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.common.utils.Bytes;
-import java.time.Duration;
import java.util.Objects;
-public class FinalResultsSuppressionBuilder<K extends Windowed> implements Suppressed<K> {
- private final StrictBufferConfig bufferConfig;
+class TimeKey implements Comparable<TimeKey> {
+ private final long time;
+ private final Bytes key;
- public FinalResultsSuppressionBuilder(final Suppressed.StrictBufferConfig bufferConfig) {
- this.bufferConfig = bufferConfig;
+ TimeKey(final long time, final Bytes key) {
+ this.time = time;
+ this.key = key;
}
- public SuppressedImpl<K> buildFinalResultsSuppression(final Duration gracePeriod) {
- return new SuppressedImpl<>(
- gracePeriod,
- bufferConfig,
- (ProcessorContext context, K key) -> key.window().end(),
- true
- );
+ Bytes key() {
+ return key;
+ }
+
+ long time() {
+ return time;
}
@Override
public boolean equals(final Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
- final FinalResultsSuppressionBuilder<?> that = (FinalResultsSuppressionBuilder<?>) o;
- return Objects.equals(bufferConfig, that.bufferConfig);
+ final TimeKey timeKey = (TimeKey) o;
+ return time == timeKey.time &&
+ Objects.equals(key, timeKey.key);
}
@Override
public int hashCode() {
- return Objects.hash(bufferConfig);
+ return Objects.hash(time, key);
}
@Override
- public String toString() {
- return "FinalResultsSuppressionBuilder{bufferConfig=" + bufferConfig + '}';
+ public int compareTo(final TimeKey o) {
+ // ordering of keys within a time uses hashCode.
+ final int timeComparison = Long.compare(time, o.time);
+ return timeComparison == 0 ? key.compareTo(o.key) : timeComparison;
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeOrderedKeyValueBuffer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeOrderedKeyValueBuffer.java
new file mode 100644
index 0000000..98a4f63
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeOrderedKeyValueBuffer.java
@@ -0,0 +1,36 @@
+/*
+ * 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.kstream.internals.suppress;
+
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.internals.ContextualRecord;
+
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+interface TimeOrderedKeyValueBuffer {
+ void evictWhile(final Supplier<Boolean> predicate, final Consumer<KeyValue<Bytes, ContextualRecord>> callback);
+
+ void put(final long time, final Bytes key, final ContextualRecord value);
+
+ int numRecords();
+
+ long bufferSize();
+
+ long minTimestamp();
+}
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
index dd57264..cd4657b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
+++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorRecordContext.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.processor.internals;
+import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.streams.processor.RecordContext;
@@ -78,6 +79,23 @@ public class ProcessorRecordContext implements RecordContext {
return headers;
}
+ public long sizeBytes() {
+ long size = 0L;
+ size += 8; // value.context.timestamp
+ size += 8; // value.context.offset
+ if (topic != null) {
+ size += topic.toCharArray().length;
+ }
+ size += 4; // partition
+ if (headers != null) {
+ for (final Header header : headers) {
+ size += header.key().toCharArray().length;
+ size += header.value().length;
+ }
+ }
+ return size;
+ }
+
@Override
public boolean equals(final Object o) {
if (this == o) return true;
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
index c016f64..a6a24ea 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
@@ -89,7 +89,7 @@ class CachingKeyValueStore<K, V> extends WrappedStateStore.AbstractStateStore im
private void putAndMaybeForward(final ThreadCache.DirtyEntry entry, final InternalProcessorContext context) {
final ProcessorRecordContext current = context.recordContext();
try {
- context.setRecordContext(entry.recordContext());
+ context.setRecordContext(entry.entry().context());
if (flushListener != null) {
V oldValue = null;
if (sendOldValues) {
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
index 2da5ab9..cbcb749 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingSessionStore.java
@@ -169,7 +169,7 @@ class CachingSessionStore<K, AGG> extends WrappedStateStore.AbstractStateStore i
private void putAndMaybeForward(final ThreadCache.DirtyEntry entry, final InternalProcessorContext context) {
final Bytes binaryKey = cacheFunction.key(entry.key());
final ProcessorRecordContext current = context.recordContext();
- context.setRecordContext(entry.recordContext());
+ context.setRecordContext(entry.entry().context());
try {
final Windowed<K> key = SessionKeySchema.from(binaryKey.get(), serdes.keyDeserializer(), topic);
final Bytes rawKey = Bytes.wrap(serdes.rawKey(key.key()));
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
index 688e889..f8d9ad5 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingWindowStore.java
@@ -108,7 +108,7 @@ class CachingWindowStore<K, V> extends WrappedStateStore.AbstractStateStore impl
final InternalProcessorContext context) {
if (flushListener != null) {
final ProcessorRecordContext current = context.recordContext();
- context.setRecordContext(entry.recordContext());
+ context.setRecordContext(entry.entry().context());
try {
final V oldValue = sendOldValues ? fetchPrevious(key, windowedKey.window().start()) : null;
flushListener.apply(windowedKey, serdes.valueFrom(entry.newValue()), oldValue);
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ContextualRecord.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ContextualRecord.java
new file mode 100644
index 0000000..89935c0
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ContextualRecord.java
@@ -0,0 +1,58 @@
+/*
+ * 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.streams.processor.internals.ProcessorRecordContext;
+
+import java.util.Arrays;
+import java.util.Objects;
+
+public class ContextualRecord {
+ private final byte[] value;
+ private final ProcessorRecordContext recordContext;
+
+ public ContextualRecord(final byte[] value, final ProcessorRecordContext recordContext) {
+ this.value = value;
+ this.recordContext = recordContext;
+ }
+
+ public ProcessorRecordContext recordContext() {
+ return recordContext;
+ }
+
+ public byte[] value() {
+ return value;
+ }
+
+ public long sizeBytes() {
+ return (value == null ? 0 : value.length) + recordContext.sizeBytes();
+ }
+
+ @Override
+ public boolean equals(final Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+ final ContextualRecord that = (ContextualRecord) o;
+ return Arrays.equals(value, that.value) &&
+ Objects.equals(recordContext, that.recordContext);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(value, recordContext);
+ }
+}
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java
index 0ac0b77..5343635 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/LRUCacheEntry.java
@@ -19,18 +19,17 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
-import java.util.Arrays;
import java.util.Objects;
/**
* A cache entry
*/
-class LRUCacheEntry extends ProcessorRecordContext {
-
- private final byte[] value;
+class LRUCacheEntry {
+ private final ContextualRecord record;
private final long sizeBytes;
private boolean isDirty;
+
LRUCacheEntry(final byte[] value) {
this(value, null, false, -1, -1, -1, "");
}
@@ -42,15 +41,16 @@ class LRUCacheEntry extends ProcessorRecordContext {
final long timestamp,
final int partition,
final String topic) {
- super(timestamp, offset, partition, topic, headers);
- this.value = value;
+ final ProcessorRecordContext context = new ProcessorRecordContext(timestamp, offset, partition, topic, headers);
+
+ this.record = new ContextualRecord(
+ value,
+ context
+ );
+
this.isDirty = isDirty;
- this.sizeBytes = (value == null ? 0 : value.length) +
- 1 + // isDirty
- 8 + // timestamp
- 8 + // offset
- 4 + // partition
- (topic == null ? 0 : topic.length());
+ this.sizeBytes = 1 + // isDirty
+ record.sizeBytes();
}
void markClean() {
@@ -66,7 +66,11 @@ class LRUCacheEntry extends ProcessorRecordContext {
}
byte[] value() {
- return value;
+ return record.value();
+ }
+
+ public ProcessorRecordContext context() {
+ return record.recordContext();
}
@Override
@@ -74,17 +78,13 @@ class LRUCacheEntry extends ProcessorRecordContext {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
final LRUCacheEntry that = (LRUCacheEntry) o;
- return timestamp() == that.timestamp() &&
- offset() == that.offset() &&
- partition() == that.partition() &&
- Objects.equals(topic(), that.topic()) &&
- Objects.equals(headers(), that.headers()) &&
- Arrays.equals(this.value, that.value()) &&
- this.isDirty == that.isDirty();
+ return sizeBytes == that.sizeBytes &&
+ isDirty() == that.isDirty() &&
+ Objects.equals(record, that.record);
}
@Override
public int hashCode() {
- return Objects.hash(timestamp(), offset(), topic(), partition(), headers(), value, isDirty);
+ return Objects.hash(record, sizeBytes, isDirty());
}
}
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
index 27270e6..941b522 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/ThreadCache.java
@@ -19,7 +19,6 @@ package org.apache.kafka.streams.state.internals;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.common.utils.LogContext;
import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.slf4j.Logger;
@@ -332,9 +331,9 @@ public class ThreadCache {
static class DirtyEntry {
private final Bytes key;
private final byte[] newValue;
- private final ProcessorRecordContext recordContext;
+ private final LRUCacheEntry recordContext;
- DirtyEntry(final Bytes key, final byte[] newValue, final ProcessorRecordContext recordContext) {
+ DirtyEntry(final Bytes key, final byte[] newValue, final LRUCacheEntry recordContext) {
this.key = key;
this.newValue = newValue;
this.recordContext = recordContext;
@@ -348,7 +347,7 @@ public class ThreadCache {
return newValue;
}
- public ProcessorRecordContext recordContext() {
+ public LRUCacheEntry entry() {
return recordContext;
}
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/SuppressionIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/SuppressionIntegrationTest.java
index af91aba..a9920e3 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/SuppressionIntegrationTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/SuppressionIntegrationTest.java
@@ -49,7 +49,6 @@ import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.test.IntegrationTest;
import org.junit.ClassRule;
-import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -75,6 +74,9 @@ import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxRecord
import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.unbounded;
import static org.apache.kafka.streams.kstream.Suppressed.untilTimeLimit;
import static org.apache.kafka.streams.kstream.Suppressed.untilWindowCloses;
+import static org.apache.kafka.test.TestUtils.waitForCondition;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
@Category({IntegrationTest.class})
public class SuppressionIntegrationTest {
@@ -88,7 +90,6 @@ public class SuppressionIntegrationTest {
private static final int SCALE_FACTOR = COMMIT_INTERVAL * 2;
private static final long TIMEOUT_MS = 30_000L;
- @Ignore
@Test
public void shouldSuppressIntermediateEventsWithEmitAfter() throws InterruptedException {
final String testId = "-shouldSuppressIntermediateEventsWithEmitAfter";
@@ -220,10 +221,9 @@ public class SuppressionIntegrationTest {
}
}
- @Ignore
@Test
public void shouldSuppressIntermediateEventsWithRecordLimit() throws InterruptedException {
- final String testId = "-shouldSuppressIntermediateEventsWithKeyLimit";
+ final String testId = "-shouldSuppressIntermediateEventsWithRecordLimit";
final String appId = getClass().getSimpleName().toLowerCase(Locale.getDefault()) + testId;
final String input = "input" + testId;
final String outputSuppressed = "output-suppressed" + testId;
@@ -279,7 +279,46 @@ public class SuppressionIntegrationTest {
}
}
- @Ignore
+ @Test
+ public void shouldShutdownWhenRecordConstraintIsViolated() throws InterruptedException {
+ final String testId = "-shouldShutdownWhenRecordConstraintIsViolated";
+ final String appId = getClass().getSimpleName().toLowerCase(Locale.getDefault()) + testId;
+ final String input = "input" + testId;
+ final String outputSuppressed = "output-suppressed" + testId;
+ final String outputRaw = "output-raw" + testId;
+
+ cleanStateBeforeTest(input, outputRaw, outputSuppressed);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ final KTable<String, Long> valueCounts = buildCountsTable(input, builder);
+
+ valueCounts
+ .suppress(untilTimeLimit(ofMillis(MAX_VALUE), maxRecords(1L).shutDownWhenFull()))
+ .toStream()
+ .to(outputSuppressed, Produced.with(STRING_SERDE, Serdes.Long()));
+
+ valueCounts
+ .toStream()
+ .to(outputRaw, Produced.with(STRING_SERDE, Serdes.Long()));
+
+ final KafkaStreams driver = getCleanStartedStreams(appId, builder);
+ try {
+ produceSynchronously(
+ input,
+ asList(
+ new KeyValueTimestamp<>("k1", "v1", scaledTime(0L)),
+ new KeyValueTimestamp<>("k1", "v2", scaledTime(1L)),
+ new KeyValueTimestamp<>("k2", "v1", scaledTime(2L)),
+ new KeyValueTimestamp<>("x", "x", scaledTime(3L))
+ )
+ );
+ verifyErrorShutdown(driver);
+ } finally {
+ driver.close();
+ cleanStateAfterTest(driver);
+ }
+ }
+
@Test
public void shouldSuppressIntermediateEventsWithBytesLimit() throws InterruptedException {
final String testId = "-shouldSuppressIntermediateEventsWithBytesLimit";
@@ -339,7 +378,47 @@ public class SuppressionIntegrationTest {
}
}
- @Ignore
+ @Test
+ public void shouldShutdownWhenBytesConstraintIsViolated() throws InterruptedException {
+ final String testId = "-shouldShutdownWhenBytesConstraintIsViolated";
+ final String appId = getClass().getSimpleName().toLowerCase(Locale.getDefault()) + testId;
+ final String input = "input" + testId;
+ final String outputSuppressed = "output-suppressed" + testId;
+ final String outputRaw = "output-raw" + testId;
+
+ cleanStateBeforeTest(input, outputRaw, outputSuppressed);
+
+ final StreamsBuilder builder = new StreamsBuilder();
+ final KTable<String, Long> valueCounts = buildCountsTable(input, builder);
+
+ valueCounts
+ // this is a bit brittle, but I happen to know that the entries are a little over 100 bytes in size.
+ .suppress(untilTimeLimit(Duration.ofMillis(MAX_VALUE), maxBytes(200L).shutDownWhenFull()))
+ .toStream()
+ .to(outputSuppressed, Produced.with(STRING_SERDE, Serdes.Long()));
+
+ valueCounts
+ .toStream()
+ .to(outputRaw, Produced.with(STRING_SERDE, Serdes.Long()));
+
+ final KafkaStreams driver = getCleanStartedStreams(appId, builder);
+ try {
+ produceSynchronously(
+ input,
+ asList(
+ new KeyValueTimestamp<>("k1", "v1", scaledTime(0L)),
+ new KeyValueTimestamp<>("k1", "v2", scaledTime(1L)),
+ new KeyValueTimestamp<>("k2", "v1", scaledTime(2L)),
+ new KeyValueTimestamp<>("x", "x", scaledTime(3L))
+ )
+ );
+ verifyErrorShutdown(driver);
+ } finally {
+ driver.close();
+ cleanStateAfterTest(driver);
+ }
+ }
+
@Test
public void shouldSupportFinalResultsForTimeWindows() throws InterruptedException {
final String testId = "-shouldSupportFinalResultsForTimeWindows";
@@ -479,6 +558,11 @@ public class SuppressionIntegrationTest {
}
}
+ private void verifyErrorShutdown(final KafkaStreams driver) throws InterruptedException {
+ waitForCondition(() -> !driver.state().isRunning(), TIMEOUT_MS, "Streams didn't shut down.");
+ assertThat(driver.state(), is(KafkaStreams.State.ERROR));
+ }
+
private void verifyOutput(final String topic, final List<KeyValueTimestamp<String, Long>> expected) {
final List<ConsumerRecord<String, Long>> results;
try {
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/SuppressedTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/SuppressedTest.java
index 7650c59..fcb5ba8 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/SuppressedTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/SuppressedTest.java
@@ -19,7 +19,7 @@ package org.apache.kafka.streams.kstream;
import org.apache.kafka.streams.kstream.internals.suppress.EagerBufferConfigImpl;
import org.apache.kafka.streams.kstream.internals.suppress.FinalResultsSuppressionBuilder;
import org.apache.kafka.streams.kstream.internals.suppress.StrictBufferConfigImpl;
-import org.apache.kafka.streams.kstream.internals.suppress.SuppressedImpl;
+import org.apache.kafka.streams.kstream.internals.suppress.SuppressedInternal;
import org.junit.Test;
import static java.lang.Long.MAX_VALUE;
@@ -61,31 +61,31 @@ public class SuppressedTest {
assertThat(
"time alone should be set",
untilTimeLimit(ofMillis(2), unbounded()),
- is(new SuppressedImpl<>(ofMillis(2), unbounded(), null, false))
+ is(new SuppressedInternal<>(ofMillis(2), unbounded(), null, false))
);
assertThat(
"time and unbounded buffer should be set",
untilTimeLimit(ofMillis(2), unbounded()),
- is(new SuppressedImpl<>(ofMillis(2), unbounded(), null, false))
+ is(new SuppressedInternal<>(ofMillis(2), unbounded(), null, false))
);
assertThat(
"time and keys buffer should be set",
untilTimeLimit(ofMillis(2), maxRecords(2)),
- is(new SuppressedImpl<>(ofMillis(2), maxRecords(2), null, false))
+ is(new SuppressedInternal<>(ofMillis(2), maxRecords(2), null, false))
);
assertThat(
"time and size buffer should be set",
untilTimeLimit(ofMillis(2), maxBytes(2)),
- is(new SuppressedImpl<>(ofMillis(2), maxBytes(2), null, false))
+ is(new SuppressedInternal<>(ofMillis(2), maxBytes(2), null, false))
);
assertThat(
"all constraints should be set",
untilTimeLimit(ofMillis(2L), maxRecords(3L).withMaxBytes(2L)),
- is(new SuppressedImpl<>(ofMillis(2), new EagerBufferConfigImpl(3L, 2L), null, false))
+ is(new SuppressedInternal<>(ofMillis(2), new EagerBufferConfigImpl(3L, 2L), null, false))
);
}
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
index d98a15e..222e1d6 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/SuppressScenarioTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams.kstream.internals;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.LongDeserializer;
@@ -32,7 +31,6 @@ import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
-import org.apache.kafka.streams.errors.ProcessorStateException;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
@@ -41,7 +39,6 @@ import org.apache.kafka.streams.kstream.Serialized;
import org.apache.kafka.streams.kstream.SessionWindows;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.kstream.internals.suppress.KTableSuppressProcessor;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.SessionStore;
import org.apache.kafka.streams.state.WindowStore;
@@ -60,6 +57,7 @@ import java.util.Properties;
import static java.time.Duration.ZERO;
import static java.time.Duration.ofMillis;
import static java.util.Arrays.asList;
+import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxBytes;
import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxRecords;
@@ -159,7 +157,7 @@ public class SuppressScenarioTest {
}
}
- @Test(expected = ProcessorStateException.class)
+ @Test
public void shouldSuppressIntermediateEventsWithTimeLimit() {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<String, Long> valueCounts = builder
@@ -198,11 +196,9 @@ public class SuppressScenarioTest {
new KeyValueTimestamp<>("v1", 1L, 2L)
)
);
- // note that the current stream time is 2, which causes v1 to age out of the buffer, since
- // it has been buffered since time 0 (even though the current version of it in the buffer has timestamp 1)
verify(
drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
- singletonList(new KeyValueTimestamp<>("v1", 0L, 1L))
+ singletonList(new KeyValueTimestamp<>("v1", 1L, 2L))
);
// inserting a dummy "tick" record just to advance stream time
driver.pipeInput(recordFactory.create("input", "tick", "tick", 3L));
@@ -225,36 +221,15 @@ public class SuppressScenarioTest {
new KeyValueTimestamp<>("tick", 1L, 4L)
)
);
+ // tick is still buffered, since it was first inserted at time 3, and it is only time 4 right now.
verify(
drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
- singletonList(
- new KeyValueTimestamp<>("v1", 1L, 2L)
- )
- );
- driver.pipeInput(recordFactory.create("input", "tick", "tick", 5L));
- verify(
- drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
- asList(
- new KeyValueTimestamp<>("tick", 0L, 5L),
- new KeyValueTimestamp<>("tick", 1L, 5L)
- )
- );
- // Note that because the punctuate runs before the process call, the tick at time 5 causes
- // the previous tick to age out of the buffer, so at this point, we see the prior value emitted
- // and the new value is still buffered.
-
- // Also worth noting is that "tick" ages out because it has been buffered since time 3, even though
- // the current timestamp of the buffered record is 4.
- verify(
- drainProducerRecords(driver, "output-suppressed", STRING_DESERIALIZER, LONG_DESERIALIZER),
- singletonList(
- new KeyValueTimestamp<>("tick", 1L, 4L)
- )
+ emptyList()
);
}
}
- @Test(expected = ProcessorStateException.class)
+ @Test
public void shouldSuppressIntermediateEventsWithRecordLimit() {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<String, Long> valueCounts = builder
@@ -320,7 +295,7 @@ public class SuppressScenarioTest {
}
}
- @Test(expected = ProcessorStateException.class)
+ @Test
public void shouldSuppressIntermediateEventsWithBytesLimit() {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<String, Long> valueCounts = builder
@@ -351,8 +326,7 @@ public class SuppressScenarioTest {
try (final TopologyTestDriver driver = new TopologyTestDriver(topology, config)) {
driver.pipeInput(recordFactory.create("input", "k1", "v1", 0L));
driver.pipeInput(recordFactory.create("input", "k1", "v2", 1L));
- final ConsumerRecord<byte[], byte[]> consumerRecord = recordFactory.create("input", "k2", "v1", 2L);
- driver.pipeInput(consumerRecord);
+ driver.pipeInput(recordFactory.create("input", "k2", "v1", 2L));
verify(
drainProducerRecords(driver, "output-raw", STRING_DESERIALIZER, LONG_DESERIALIZER),
asList(
@@ -388,7 +362,7 @@ public class SuppressScenarioTest {
}
}
- @Test(expected = KTableSuppressProcessor.NotImplementedException.class)
+ @Test
public void shouldSupportFinalResultsForTimeWindows() {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<Windowed<String>, Long> valueCounts = builder
@@ -442,7 +416,7 @@ public class SuppressScenarioTest {
}
}
- @Test(expected = KTableSuppressProcessor.NotImplementedException.class)
+ @Test
public void shouldSupportFinalResultsForTimeWindowsWithLargeJump() {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<Windowed<String>, Long> valueCounts = builder
@@ -501,7 +475,7 @@ public class SuppressScenarioTest {
}
}
- @Test(expected = KTableSuppressProcessor.NotImplementedException.class)
+ @Test
public void shouldSupportFinalResultsForSessionWindows() {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<Windowed<String>, Long> valueCounts = builder
diff --git a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java
index a38d1d5..bb7f49c 100644
--- a/streams/src/test/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/kstream/internals/suppress/KTableSuppressProcessorTest.java
@@ -16,15 +16,20 @@
*/
package org.apache.kafka.streams.kstream.internals.suppress;
+import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.errors.StreamsException;
import org.apache.kafka.streams.kstream.Suppressed;
+import org.apache.kafka.streams.kstream.TimeWindowedDeserializer;
+import org.apache.kafka.streams.kstream.TimeWindowedSerializer;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.Change;
import org.apache.kafka.streams.kstream.internals.FullChangeSerde;
import org.apache.kafka.streams.kstream.internals.SessionWindow;
import org.apache.kafka.streams.kstream.internals.TimeWindow;
import org.apache.kafka.streams.processor.MockProcessorContext;
+import org.apache.kafka.streams.processor.internals.ProcessorNode;
import org.apache.kafka.test.MockInternalProcessorContext;
import org.hamcrest.BaseMatcher;
import org.hamcrest.Description;
@@ -38,25 +43,23 @@ import static java.time.Duration.ZERO;
import static java.time.Duration.ofMillis;
import static org.apache.kafka.common.serialization.Serdes.Long;
import static org.apache.kafka.common.serialization.Serdes.String;
+import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxBytes;
import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.maxRecords;
import static org.apache.kafka.streams.kstream.Suppressed.BufferConfig.unbounded;
import static org.apache.kafka.streams.kstream.Suppressed.untilTimeLimit;
import static org.apache.kafka.streams.kstream.Suppressed.untilWindowCloses;
import static org.apache.kafka.streams.kstream.WindowedSerdes.sessionWindowedSerdeFrom;
-import static org.apache.kafka.streams.kstream.WindowedSerdes.timeWindowedSerdeFrom;
+import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
@SuppressWarnings("PointlessArithmeticExpression")
public class KTableSuppressProcessorTest {
private static final long ARBITRARY_LONG = 5L;
- private static final long ARBITRARY_TIMESTAMP = 1993L;
-
private static final Change<Long> ARBITRARY_CHANGE = new Change<>(7L, 14L);
- private static final TimeWindow ARBITRARY_WINDOW = new TimeWindow(0L, 100L);
-
@Test
public void zeroTimeLimitShouldImmediatelyEmit() {
final KTableSuppressProcessor<String, Long> processor =
@@ -66,7 +69,7 @@ public class KTableSuppressProcessorTest {
processor.init(context);
final long timestamp = ARBITRARY_LONG;
- context.setTimestamp(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
context.setStreamTime(timestamp);
final String key = "hey";
final Change<Long> value = ARBITRARY_CHANGE;
@@ -83,7 +86,7 @@ public class KTableSuppressProcessorTest {
final KTableSuppressProcessor<Windowed<String>, Long> processor =
new KTableSuppressProcessor<>(
getImpl(untilTimeLimit(ZERO, unbounded())),
- timeWindowedSerdeFrom(String.class),
+ timeWindowedSerdeFrom(String.class, 100L),
new FullChangeSerde<>(Long())
);
@@ -91,9 +94,9 @@ public class KTableSuppressProcessorTest {
processor.init(context);
final long timestamp = ARBITRARY_LONG;
- context.setTimestamp(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
context.setStreamTime(timestamp);
- final Windowed<String> key = new Windowed<>("hey", ARBITRARY_WINDOW);
+ final Windowed<String> key = new Windowed<>("hey", new TimeWindow(0L, 100L));
final Change<Long> value = ARBITRARY_CHANGE;
processor.process(key, value);
@@ -103,7 +106,7 @@ public class KTableSuppressProcessorTest {
assertThat(capturedForward.timestamp(), is(timestamp));
}
- @Test(expected = KTableSuppressProcessor.NotImplementedException.class)
+ @Test
public void intermediateSuppressionShouldBufferAndEmitLater() {
final KTableSuppressProcessor<String, Long> processor =
new KTableSuppressProcessor<>(
@@ -117,13 +120,15 @@ public class KTableSuppressProcessorTest {
final long timestamp = 0L;
context.setRecordMetadata("topic", 0, 0, null, timestamp);
+ context.setStreamTime(timestamp);
final String key = "hey";
final Change<Long> value = new Change<>(null, 1L);
processor.process(key, value);
assertThat(context.forwarded(), hasSize(0));
- assertThat(context.scheduledPunctuators(), hasSize(1));
- context.scheduledPunctuators().get(0).getPunctuator().punctuate(1);
+ context.setRecordMetadata("topic", 0, 1, null, 1L);
+ context.setStreamTime(1L);
+ processor.process("tick", new Change<>(null, null));
assertThat(context.forwarded(), hasSize(1));
final MockProcessorContext.CapturedForward capturedForward = context.forwarded().get(0);
@@ -131,38 +136,49 @@ public class KTableSuppressProcessorTest {
assertThat(capturedForward.timestamp(), is(timestamp));
}
-
- @SuppressWarnings("unchecked")
- private <K extends Windowed> SuppressedImpl<K> finalResults(final Duration grace) {
- return ((FinalResultsSuppressionBuilder) untilWindowCloses(unbounded())).buildFinalResultsSuppression(grace);
- }
-
-
- @Test(expected = KTableSuppressProcessor.NotImplementedException.class)
+ @Test
public void finalResultsSuppressionShouldBufferAndEmitAtGraceExpiration() {
final KTableSuppressProcessor<Windowed<String>, Long> processor = new KTableSuppressProcessor<>(
finalResults(ofMillis(1L)),
- timeWindowedSerdeFrom(String.class),
+ timeWindowedSerdeFrom(String.class, 1L),
new FullChangeSerde<>(Long())
);
final MockInternalProcessorContext context = new MockInternalProcessorContext();
processor.init(context);
- final long timestamp = ARBITRARY_TIMESTAMP;
- context.setRecordMetadata("topic", 0, 0, null, timestamp);
- final Windowed<String> key = new Windowed<>("hey", ARBITRARY_WINDOW);
+ final long windowStart = 99L;
+ final long recordTime = 99L;
+ final long windowEnd = 100L;
+ context.setRecordMetadata("topic", 0, 0, null, recordTime);
+ context.setStreamTime(recordTime);
+ final Windowed<String> key = new Windowed<>("hey", new TimeWindow(windowStart, windowEnd));
final Change<Long> value = ARBITRARY_CHANGE;
processor.process(key, value);
assertThat(context.forwarded(), hasSize(0));
- assertThat(context.scheduledPunctuators(), hasSize(1));
- context.scheduledPunctuators().get(0).getPunctuator().punctuate(timestamp + 1L);
+ // although the stream time is now 100, we have to wait 1 ms after the window *end* before we
+ // emit "hey", so we don't emit yet.
+ final long windowStart2 = 100L;
+ final long recordTime2 = 100L;
+ final long windowEnd2 = 101L;
+ context.setRecordMetadata("topic", 0, 1, null, recordTime2);
+ context.setStreamTime(recordTime2);
+ processor.process(new Windowed<>("dummyKey1", new TimeWindow(windowStart2, windowEnd2)), ARBITRARY_CHANGE);
+ assertThat(context.forwarded(), hasSize(0));
+
+ // ok, now it's time to emit "hey"
+ final long windowStart3 = 101L;
+ final long recordTime3 = 101L;
+ final long windowEnd3 = 102L;
+ context.setRecordMetadata("topic", 0, 1, null, recordTime3);
+ context.setStreamTime(recordTime3);
+ processor.process(new Windowed<>("dummyKey2", new TimeWindow(windowStart3, windowEnd3)), ARBITRARY_CHANGE);
assertThat(context.forwarded(), hasSize(1));
final MockProcessorContext.CapturedForward capturedForward = context.forwarded().get(0);
assertThat(capturedForward.keyValue(), is(new KeyValue<>(key, value)));
- assertThat(capturedForward.timestamp(), is(timestamp));
+ assertThat(capturedForward.timestamp(), is(recordTime));
}
/**
@@ -170,27 +186,32 @@ public class KTableSuppressProcessorTest {
* it will still buffer events and emit only after the end of the window.
* As opposed to emitting immediately the way regular suppresion would with a time limit of 0.
*/
- @Test(expected = KTableSuppressProcessor.NotImplementedException.class)
+ @Test
public void finalResultsWithZeroGraceShouldStillBufferUntilTheWindowEnd() {
final KTableSuppressProcessor<Windowed<String>, Long> processor = new KTableSuppressProcessor<>(
finalResults(ofMillis(0)),
- timeWindowedSerdeFrom(String.class),
+ timeWindowedSerdeFrom(String.class, 100L),
new FullChangeSerde<>(Long())
);
final MockInternalProcessorContext context = new MockInternalProcessorContext();
processor.init(context);
+ // note the record is in the past, but the window end is in the future, so we still have to buffer,
+ // even though the grace period is 0.
final long timestamp = 5L;
- context.setRecordMetadata("", 0, 0L, null, timestamp);
+ final long streamTime = 99L;
final long windowEnd = 100L;
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
+ context.setStreamTime(streamTime);
final Windowed<String> key = new Windowed<>("hey", new TimeWindow(0, windowEnd));
final Change<Long> value = ARBITRARY_CHANGE;
processor.process(key, value);
assertThat(context.forwarded(), hasSize(0));
- assertThat(context.scheduledPunctuators(), hasSize(1));
- context.scheduledPunctuators().get(0).getPunctuator().punctuate(windowEnd);
+ context.setRecordMetadata("", 0, 1L, null, windowEnd);
+ context.setStreamTime(windowEnd);
+ processor.process(new Windowed<>("dummyKey", new TimeWindow(windowEnd, windowEnd + 100L)), ARBITRARY_CHANGE);
assertThat(context.forwarded(), hasSize(1));
final MockProcessorContext.CapturedForward capturedForward = context.forwarded().get(0);
@@ -202,7 +223,7 @@ public class KTableSuppressProcessorTest {
public void finalResultsWithZeroGraceAtWindowEndShouldImmediatelyEmit() {
final KTableSuppressProcessor<Windowed<String>, Long> processor = new KTableSuppressProcessor<>(
finalResults(ofMillis(0)),
- timeWindowedSerdeFrom(String.class),
+ timeWindowedSerdeFrom(String.class, 100L),
new FullChangeSerde<>(Long())
);
@@ -210,7 +231,7 @@ public class KTableSuppressProcessorTest {
processor.init(context);
final long timestamp = 100L;
- context.setTimestamp(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
context.setStreamTime(timestamp);
final Windowed<String> key = new Windowed<>("hey", new TimeWindow(0, 100L));
final Change<Long> value = ARBITRARY_CHANGE;
@@ -226,7 +247,7 @@ public class KTableSuppressProcessorTest {
public void finalResultsShouldSuppressTombstonesForTimeWindows() {
final KTableSuppressProcessor<Windowed<String>, Long> processor = new KTableSuppressProcessor<>(
finalResults(ofMillis(0)),
- timeWindowedSerdeFrom(String.class),
+ timeWindowedSerdeFrom(String.class, 100L),
new FullChangeSerde<>(Long())
);
@@ -234,7 +255,7 @@ public class KTableSuppressProcessorTest {
processor.init(context);
final long timestamp = 100L;
- context.setTimestamp(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
context.setStreamTime(timestamp);
final Windowed<String> key = new Windowed<>("hey", new TimeWindow(0, 100L));
final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
@@ -255,7 +276,7 @@ public class KTableSuppressProcessorTest {
processor.init(context);
final long timestamp = 100L;
- context.setTimestamp(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
context.setStreamTime(timestamp);
final Windowed<String> key = new Windowed<>("hey", new SessionWindow(0L, 0L));
final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
@@ -264,12 +285,11 @@ public class KTableSuppressProcessorTest {
assertThat(context.forwarded(), hasSize(0));
}
- @SuppressWarnings("unchecked")
@Test
public void suppressShouldNotSuppressTombstonesForTimeWindows() {
- final KTableSuppressProcessor<Windowed<String>, Long> processor = new KTableSuppressProcessor<Windowed<String>, Long>(
- (SuppressedImpl) untilTimeLimit(ofMillis(0), maxRecords(0)),
- timeWindowedSerdeFrom(String.class),
+ final KTableSuppressProcessor<Windowed<String>, Long> processor = new KTableSuppressProcessor<>(
+ getImpl(untilTimeLimit(ofMillis(0), maxRecords(0))),
+ timeWindowedSerdeFrom(String.class, 100L),
new FullChangeSerde<>(Long())
);
@@ -277,7 +297,7 @@ public class KTableSuppressProcessorTest {
processor.init(context);
final long timestamp = 100L;
- context.setTimestamp(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
context.setStreamTime(timestamp);
final Windowed<String> key = new Windowed<>("hey", new TimeWindow(0L, 100L));
final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
@@ -289,11 +309,10 @@ public class KTableSuppressProcessorTest {
assertThat(capturedForward.timestamp(), is(timestamp));
}
- @SuppressWarnings("unchecked")
@Test
public void suppressShouldNotSuppressTombstonesForSessionWindows() {
- final KTableSuppressProcessor<Windowed<String>, Long> processor = new KTableSuppressProcessor<Windowed<String>, Long>(
- (SuppressedImpl) untilTimeLimit(ofMillis(0), maxRecords(0)),
+ final KTableSuppressProcessor<Windowed<String>, Long> processor = new KTableSuppressProcessor<>(
+ getImpl(untilTimeLimit(ofMillis(0), maxRecords(0))),
sessionWindowedSerdeFrom(String.class),
new FullChangeSerde<>(Long())
);
@@ -302,7 +321,7 @@ public class KTableSuppressProcessorTest {
processor.init(context);
final long timestamp = 100L;
- context.setTimestamp(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
context.setStreamTime(timestamp);
final Windowed<String> key = new Windowed<>("hey", new SessionWindow(0L, 0L));
final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
@@ -314,11 +333,61 @@ public class KTableSuppressProcessorTest {
assertThat(capturedForward.timestamp(), is(timestamp));
}
- @SuppressWarnings("unchecked")
@Test
public void suppressShouldNotSuppressTombstonesForKTable() {
- final KTableSuppressProcessor<String, Long> processor = new KTableSuppressProcessor<String, Long>(
- (SuppressedImpl) untilTimeLimit(ofMillis(0), maxRecords(0)),
+ final KTableSuppressProcessor<String, Long> processor = new KTableSuppressProcessor<>(
+ getImpl(untilTimeLimit(ofMillis(0), maxRecords(0))),
+ Serdes.String(),
+ new FullChangeSerde<>(Long())
+ );
+
+ final MockInternalProcessorContext context = new MockInternalProcessorContext();
+ processor.init(context);
+
+ final long timestamp = 100L;
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
+ context.setStreamTime(timestamp);
+ final String key = "hey";
+ final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
+ processor.process(key, value);
+
+ assertThat(context.forwarded(), hasSize(1));
+ final MockProcessorContext.CapturedForward capturedForward = context.forwarded().get(0);
+ assertThat(capturedForward.keyValue(), is(new KeyValue<>(key, value)));
+ assertThat(capturedForward.timestamp(), is(timestamp));
+ }
+
+ @Test
+ public void suppressShouldEmitWhenOverRecordCapacity() {
+ final KTableSuppressProcessor<String, Long> processor = new KTableSuppressProcessor<>(
+ getImpl(untilTimeLimit(Duration.ofDays(100), maxRecords(1))),
+ Serdes.String(),
+ new FullChangeSerde<>(Long())
+ );
+
+ final MockInternalProcessorContext context = new MockInternalProcessorContext();
+ processor.init(context);
+
+ final long timestamp = 100L;
+ context.setStreamTime(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
+ final String key = "hey";
+ final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
+ processor.process(key, value);
+
+ context.setRecordMetadata("", 0, 1L, null, timestamp + 1);
+ processor.process("dummyKey", value);
+
+ assertThat(context.forwarded(), hasSize(1));
+ final MockProcessorContext.CapturedForward capturedForward = context.forwarded().get(0);
+ assertThat(capturedForward.keyValue(), is(new KeyValue<>(key, value)));
+ assertThat(capturedForward.timestamp(), is(timestamp));
+ }
+
+ @Test
+ public void suppressShouldEmitWhenOverByteCapacity() {
+ final KTableSuppressProcessor<String, Long> processor = new KTableSuppressProcessor<>(
+ getImpl(untilTimeLimit(Duration.ofDays(100), maxBytes(60L))),
Serdes.String(),
new FullChangeSerde<>(Long())
);
@@ -327,18 +396,82 @@ public class KTableSuppressProcessorTest {
processor.init(context);
final long timestamp = 100L;
- context.setTimestamp(timestamp);
context.setStreamTime(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
final String key = "hey";
final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
processor.process(key, value);
+ context.setRecordMetadata("", 0, 1L, null, timestamp + 1);
+ processor.process("dummyKey", value);
+
assertThat(context.forwarded(), hasSize(1));
final MockProcessorContext.CapturedForward capturedForward = context.forwarded().get(0);
assertThat(capturedForward.keyValue(), is(new KeyValue<>(key, value)));
assertThat(capturedForward.timestamp(), is(timestamp));
}
+ @Test
+ public void suppressShouldShutDownWhenOverRecordCapacity() {
+ final KTableSuppressProcessor<String, Long> processor = new KTableSuppressProcessor<>(
+ getImpl(untilTimeLimit(Duration.ofDays(100), maxRecords(1).shutDownWhenFull())),
+ Serdes.String(),
+ new FullChangeSerde<>(Long())
+ );
+
+ final MockInternalProcessorContext context = new MockInternalProcessorContext();
+ processor.init(context);
+
+ final long timestamp = 100L;
+ context.setStreamTime(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
+ context.setCurrentNode(new ProcessorNode("testNode"));
+ final String key = "hey";
+ final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
+ processor.process(key, value);
+
+ context.setRecordMetadata("", 0, 1L, null, timestamp);
+ try {
+ processor.process("dummyKey", value);
+ fail("expected an exception");
+ } catch (final StreamsException e) {
+ assertThat(e.getMessage(), containsString("buffer exceeded its max capacity"));
+ }
+ }
+
+ @Test
+ public void suppressShouldShutDownWhenOverByteCapacity() {
+ final KTableSuppressProcessor<String, Long> processor = new KTableSuppressProcessor<>(
+ getImpl(untilTimeLimit(Duration.ofDays(100), maxBytes(60L).shutDownWhenFull())),
+ Serdes.String(),
+ new FullChangeSerde<>(Long())
+ );
+
+ final MockInternalProcessorContext context = new MockInternalProcessorContext();
+ processor.init(context);
+
+ final long timestamp = 100L;
+ context.setStreamTime(timestamp);
+ context.setRecordMetadata("", 0, 0L, null, timestamp);
+ context.setCurrentNode(new ProcessorNode("testNode"));
+ final String key = "hey";
+ final Change<Long> value = new Change<>(null, ARBITRARY_LONG);
+ processor.process(key, value);
+
+ context.setRecordMetadata("", 0, 1L, null, timestamp);
+ try {
+ processor.process("dummyKey", value);
+ fail("expected an exception");
+ } catch (final StreamsException e) {
+ assertThat(e.getMessage(), containsString("buffer exceeded its max capacity"));
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private <K extends Windowed> SuppressedInternal<K> finalResults(final Duration grace) {
+ return ((FinalResultsSuppressionBuilder) untilWindowCloses(unbounded())).buildFinalResultsSuppression(grace);
+ }
+
private static <E> Matcher<Collection<E>> hasSize(final int i) {
return new BaseMatcher<Collection<E>>() {
@Override
@@ -359,7 +492,15 @@ public class KTableSuppressProcessorTest {
};
}
- private static <K> SuppressedImpl<K> getImpl(final Suppressed<K> suppressed) {
- return (SuppressedImpl<K>) suppressed;
+ private static <K> SuppressedInternal<K> getImpl(final Suppressed<K> suppressed) {
+ return (SuppressedInternal<K>) suppressed;
+ }
+
+ private <K> Serde<Windowed<K>> timeWindowedSerdeFrom(final Class<K> rawType, final long windowSize) {
+ final Serde<K> kSerde = Serdes.serdeFrom(rawType);
+ return new Serdes.WrapperSerde<>(
+ new TimeWindowedSerializer<>(kSerde.serializer()),
+ new TimeWindowedDeserializer<>(kSerde.deserializer(), windowSize)
+ );
}
}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java
index 0fdbdf7..71a6ac2 100644
--- a/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/NamedCacheTest.java
@@ -190,7 +190,7 @@ public class NamedCacheTest {
assertEquals(2, flushed.size());
assertEquals(Bytes.wrap(new byte[] {0}), flushed.get(0).key());
- assertEquals(headers, flushed.get(0).recordContext().headers());
+ assertEquals(headers, flushed.get(0).entry().context().headers());
assertArrayEquals(new byte[] {10}, flushed.get(0).newValue());
assertEquals(Bytes.wrap(new byte[] {2}), flushed.get(1).key());
assertArrayEquals(new byte[] {30}, flushed.get(1).newValue());