[
https://issues.apache.org/jira/browse/KAFKA-7223?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16635064#comment-16635064
]
ASF GitHub Bot commented on KAFKA-7223:
---------------------------------------
mjsax closed pull request #5693: KAFKA-7223: In-Memory Suppression Buffering
URL: https://github.com/apache/kafka/pull/5693
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
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 49fe96ba20c..6db7a708541 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 @@
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 @@ static StrictBufferConfig unbounded() {
* @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 8a2e619b7b5..9bb83733c82 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 @@
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 void close() {
@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 void configure(final Map<String, ?> configs, final
boolean isKey) {
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 void configure(final Map<String, ?> configs, final
boolean isKey) {
if (newBytes != null) {
buffer.put(newBytes);
}
- return byteBufferSerializer.serialize(null, buffer);
+ return buffer.array();
}
@Override
@@ -89,7 +97,6 @@ public void close() {
@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 void configure(final Map<String, ?> configs, final
boolean isKey) {
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 00000000000..a69002f9900
--- /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 c5b29702c7c..9c76766d9c8 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
@@ -39,7 +39,7 @@
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;
@@ -356,12 +356,11 @@ public String queryableStoreName() {
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<>(
@@ -387,18 +386,18 @@ public String queryableStoreName() {
}
@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 2ee8f7c5958..753ef0f85a8 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.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 @@
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 @@
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 @@
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 e731dc6f5e1..67d3783867c 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 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 @@
@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 0c2c883e18a..161f934f3a2 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 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 EagerBufferConfigImpl(final long maxKeys, final long
maxBytes) {
@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 boolean equals(final Object o) {
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 db09307d48c..523ae0602c6 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 @@
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 FinalResultsSuppressionBuilder(final
Suppressed.StrictBufferConfig buffer
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 00000000000..677a662f79d
--- /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 6f0021fbc49..57e5066d09e 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 0634a748a5b..ef754ec6fc2 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 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 StrictBufferConfigImpl() {
@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 boolean equals(final Object o) {
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 a3bf2db63a2..99245dae544 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 @@ Duration getTimeToWaitForMoreEvents() {
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 int hashCode() {
@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 00000000000..b37bcf663fd
--- /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/TimeKey.java
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeKey.java
new file mode 100644
index 00000000000..d3ad350686a
--- /dev/null
+++
b/streams/src/main/java/org/apache/kafka/streams/kstream/internals/suppress/TimeKey.java
@@ -0,0 +1,60 @@
+/*
+ * 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 java.util.Objects;
+
+class TimeKey implements Comparable<TimeKey> {
+ private final long time;
+ private final Bytes key;
+
+ TimeKey(final long time, final Bytes key) {
+ this.time = time;
+ this.key = key;
+ }
+
+ 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 TimeKey timeKey = (TimeKey) o;
+ return time == timeKey.time &&
+ Objects.equals(key, timeKey.key);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(time, key);
+ }
+
+ @Override
+ 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 00000000000..98a4f63c83f
--- /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 dd572649765..cd4657bdcd8 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 Headers headers() {
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 c016f640cb2..a6a24ea098f 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 @@ public void apply(final List<ThreadCache.DirtyEntry> entries)
{
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 2da5ab98550..cbcb7490efb 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 @@ public void put(final Windowed<Bytes> key, final byte[]
value) {
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 688e88962a2..f8d9ad590a2 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 @@ private void maybeForward(final ThreadCache.DirtyEntry
entry,
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 00000000000..89935c09d8f
--- /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 0ac0b77dd37..53436358ded 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 @@
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 @@
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 @@ long size() {
}
byte[] value() {
- return value;
+ return record.value();
+ }
+
+ public ProcessorRecordContext context() {
+ return record.recordContext();
}
@Override
@@ -74,17 +78,13 @@ public boolean equals(final Object o) {
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 27270e6b51d..941b5221524 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 @@
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 void close() {
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 Bytes key() {
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 af91abaf2b1..a9920e3a6f1 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.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.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 @@
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 void
shouldNotSuppressIntermediateEventsWithZeroEmitAfter() throws Interr
}
}
- @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 void
shouldSuppressIntermediateEventsWithRecordLimit() throws Interrupted
}
}
- @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 void
shouldSuppressIntermediateEventsWithBytesLimit() throws InterruptedE
}
}
- @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 @@ private void produceSynchronously(final String topic,
final List<KeyValueTimesta
}
}
+ 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 7650c59759e..fcb5ba8ef3f 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 @@
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 void
intermediateEventsShouldAcceptAnyBufferAndSetBounds() {
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 d98a15e093b..222e1d63982 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.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.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 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 void shouldImmediatelyEmitEventsWithZeroEmitAfter() {
}
}
- @Test(expected = ProcessorStateException.class)
+ @Test
public void shouldSuppressIntermediateEventsWithTimeLimit() {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<String, Long> valueCounts = builder
@@ -198,11 +196,9 @@ public void
shouldSuppressIntermediateEventsWithTimeLimit() {
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 void
shouldSuppressIntermediateEventsWithTimeLimit() {
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 void
shouldSuppressIntermediateEventsWithRecordLimit() {
}
}
- @Test(expected = ProcessorStateException.class)
+ @Test
public void shouldSuppressIntermediateEventsWithBytesLimit() {
final StreamsBuilder builder = new StreamsBuilder();
final KTable<String, Long> valueCounts = builder
@@ -351,8 +326,7 @@ public void
shouldSuppressIntermediateEventsWithBytesLimit() {
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 void
shouldSuppressIntermediateEventsWithBytesLimit() {
}
}
- @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 void shouldSupportFinalResultsForTimeWindows() {
}
}
- @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 void
shouldSupportFinalResultsForTimeWindowsWithLargeJump() {
}
}
- @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 a38d1d58f43..bb7f49ce7a5 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.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 void zeroTimeLimitShouldImmediatelyEmit() {
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 void windowedZeroTimeLimitShouldImmediatelyEmit() {
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 void windowedZeroTimeLimitShouldImmediatelyEmit() {
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 void windowedZeroTimeLimitShouldImmediatelyEmit() {
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 void
intermediateSuppressionShouldBufferAndEmitLater() {
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 void
intermediateSuppressionShouldBufferAndEmitLater() {
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 void
finalResultsSuppressionShouldBufferAndEmitAtGraceExpiration() {
* 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 void
finalResultsWithZeroGraceShouldStillBufferUntilTheWindowEnd() {
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 void
finalResultsWithZeroGraceAtWindowEndShouldImmediatelyEmit() {
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 void
finalResultsWithZeroGraceAtWindowEndShouldImmediatelyEmit() {
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 void
finalResultsShouldSuppressTombstonesForTimeWindows() {
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 void
finalResultsShouldSuppressTombstonesForSessionWindows() {
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 void
finalResultsShouldSuppressTombstonesForSessionWindows() {
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 void
suppressShouldNotSuppressTombstonesForTimeWindows() {
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 void
suppressShouldNotSuppressTombstonesForTimeWindows() {
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 void
suppressShouldNotSuppressTombstonesForSessionWindows() {
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 void
suppressShouldNotSuppressTombstonesForSessionWindows() {
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 void
suppressShouldNotSuppressTombstonesForKTable() {
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 boolean matches(final Object item) {
};
}
- 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 0fdbdf76b3c..71a6ac21c10 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 void apply(final List<ThreadCache.DirtyEntry> dirty)
{
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());
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> KIP-328: Add in-memory Suppression
> ----------------------------------
>
> Key: KAFKA-7223
> URL: https://issues.apache.org/jira/browse/KAFKA-7223
> Project: Kafka
> Issue Type: Improvement
> Components: streams
> Reporter: John Roesler
> Assignee: John Roesler
> Priority: Major
>
> As described in
> [https://cwiki.apache.org/confluence/display/KAFKA/KIP-328%3A+Ability+to+suppress+updates+for+KTables.]
>
> This ticket is to implement Suppress, but only for in-memory buffers.
> (depends on KAFKA-7222)
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
