You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by wc...@apache.org on 2023/06/14 20:17:22 UTC
[kafka] branch trunk updated: KAFKA-14936: Add On Disk Time Ordered Buffer (1/N) (#13756)
This is an automated email from the ASF dual-hosted git repository.
wcarlson 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 4a5d1b32050 KAFKA-14936: Add On Disk Time Ordered Buffer (1/N) (#13756)
4a5d1b32050 is described below
commit 4a5d1b320502c808509e9b795bd669030ce03fd2
Author: Walker Carlson <18...@users.noreply.github.com>
AuthorDate: Wed Jun 14 15:16:55 2023 -0500
KAFKA-14936: Add On Disk Time Ordered Buffer (1/N) (#13756)
KAFKA-14936: Add On Disk Time Ordered Buffer
Add a time ordered key-value buffer stored on disk and implemented using RocksDBTimeOrderedKeyValueSegmentedBytesStore.
This will be used in the stream buffered for joins with a grace period.
Reviewers: Bruno Cadonna <ca...@confluent.io> Victoria Xia <vi...@confluent.io>
---
.../org/apache/kafka/streams/state/Stores.java | 6 +-
...java => RocksDBKeyValueBytesStoreSupplier.java} | 4 +-
.../RocksDBTimeOrderedKeyValueBuffer.java | 174 +++++++++++++++++++
.../RocksDBTimeOrderedKeyValueBytesStore.java | 52 ++++++
...ksDBTimeOrderedKeyValueBytesStoreSupplier.java} | 30 ++--
.../RocksDBTimeOrderedKeyValueBufferTest.java | 191 +++++++++++++++++++++
6 files changed, 434 insertions(+), 23 deletions(-)
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
index 02805775bff..d3fc1fb00eb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
@@ -24,7 +24,7 @@ import org.apache.kafka.streams.state.internals.InMemorySessionBytesStoreSupplie
import org.apache.kafka.streams.state.internals.InMemoryWindowBytesStoreSupplier;
import org.apache.kafka.streams.state.internals.KeyValueStoreBuilder;
import org.apache.kafka.streams.state.internals.MemoryNavigableLRUCache;
-import org.apache.kafka.streams.state.internals.RocksDbKeyValueBytesStoreSupplier;
+import org.apache.kafka.streams.state.internals.RocksDBKeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.internals.RocksDbSessionBytesStoreSupplier;
import org.apache.kafka.streams.state.internals.RocksDbVersionedKeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.internals.RocksDbWindowBytesStoreSupplier;
@@ -94,7 +94,7 @@ public final class Stores {
*/
public static KeyValueBytesStoreSupplier persistentKeyValueStore(final String name) {
Objects.requireNonNull(name, "name cannot be null");
- return new RocksDbKeyValueBytesStoreSupplier(name, false);
+ return new RocksDBKeyValueBytesStoreSupplier(name, false);
}
/**
@@ -113,7 +113,7 @@ public final class Stores {
*/
public static KeyValueBytesStoreSupplier persistentTimestampedKeyValueStore(final String name) {
Objects.requireNonNull(name, "name cannot be null");
- return new RocksDbKeyValueBytesStoreSupplier(name, true);
+ return new RocksDBKeyValueBytesStoreSupplier(name, true);
}
/**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbKeyValueBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueBytesStoreSupplier.java
similarity index 92%
copy from streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbKeyValueBytesStoreSupplier.java
copy to streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueBytesStoreSupplier.java
index 87be76781fe..6f5d6d547db 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbKeyValueBytesStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBKeyValueBytesStoreSupplier.java
@@ -20,12 +20,12 @@ import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
import org.apache.kafka.streams.state.KeyValueStore;
-public class RocksDbKeyValueBytesStoreSupplier implements KeyValueBytesStoreSupplier {
+public class RocksDBKeyValueBytesStoreSupplier implements KeyValueBytesStoreSupplier {
private final String name;
private final boolean returnTimestampedStore;
- public RocksDbKeyValueBytesStoreSupplier(final String name,
+ public RocksDBKeyValueBytesStoreSupplier(final String name,
final boolean returnTimestampedStore) {
this.name = name;
this.returnTimestampedStore = returnTimestampedStore;
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java
new file mode 100644
index 00000000000..04aa3a56ef9
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java
@@ -0,0 +1,174 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals;
+
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.processor.ProcessorContext;
+import org.apache.kafka.streams.processor.StateStore;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.processor.internals.SerdeGetter;
+import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.ValueAndTimestamp;
+
+import java.nio.ByteBuffer;
+import java.time.Duration;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+
+import static java.util.Objects.requireNonNull;
+
+public class RocksDBTimeOrderedKeyValueBuffer<K, V> extends WrappedStateStore<RocksDBTimeOrderedKeyValueBytesStore, Object, Object> implements TimeOrderedKeyValueBuffer<K, V, V> {
+
+ private final long gracePeriod;
+ private long bufferSize;
+ private long minTimestamp;
+ private int numRecords;
+ private Serde<K> keySerde;
+ private Serde<V> valueSerde;
+ private final String topic;
+ private int seqnum;
+
+ public RocksDBTimeOrderedKeyValueBuffer(final RocksDBTimeOrderedKeyValueBytesStore store,
+ final Duration gracePeriod,
+ final String topic) {
+ super(store);
+ this.gracePeriod = gracePeriod.toMillis();
+ minTimestamp = Long.MAX_VALUE;
+ numRecords = 0;
+ bufferSize = 0;
+ seqnum = 0;
+ this.topic = topic;
+ }
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public void setSerdesIfNull(final SerdeGetter getter) {
+ keySerde = keySerde == null ? (Serde<K>) getter.keySerde() : keySerde;
+ valueSerde = valueSerde == null ? getter.valueSerde() : valueSerde;
+ }
+
+ @Deprecated
+ @Override
+ public void init(final ProcessorContext context, final StateStore root) {
+ wrapped().init(context, wrapped());
+ }
+
+ @Override
+ public void init(final StateStoreContext context, final StateStore root) {
+ wrapped().init(context, wrapped());
+ }
+
+ @Override
+ public void evictWhile(final Supplier<Boolean> predicate, final Consumer<Eviction<K, V>> callback) {
+ KeyValue<Bytes, byte[]> keyValue;
+
+ if (predicate.get()) {
+ try (final KeyValueIterator<Bytes, byte[]> iterator = wrapped()
+ .fetchAll(0, wrapped().observedStreamTime - gracePeriod)) {
+ while (iterator.hasNext() && predicate.get()) {
+ keyValue = iterator.next();
+
+ final BufferValue bufferValue = BufferValue.deserialize(ByteBuffer.wrap(keyValue.value));
+ final K key = keySerde.deserializer().deserialize(topic,
+ PrefixedWindowKeySchemas.TimeFirstWindowKeySchema.extractStoreKeyBytes(keyValue.key.get()));
+
+ if (bufferValue.context().timestamp() < minTimestamp) {
+ throw new IllegalStateException(
+ "minTimestamp [" + minTimestamp + "] did not match the actual min timestamp [" +
+ bufferValue.context().timestamp() + "]"
+ );
+ }
+ minTimestamp = bufferValue.context().timestamp();
+
+ final V value = valueSerde.deserializer().deserialize(topic, bufferValue.newValue());
+
+ callback.accept(new Eviction<>(key, value, bufferValue.context()));
+
+ wrapped().remove(keyValue.key);
+ numRecords--;
+ bufferSize = bufferSize - computeRecordSize(keyValue.key, bufferValue);
+ }
+ if (numRecords == 0) {
+ minTimestamp = Long.MAX_VALUE;
+ } else {
+ minTimestamp = wrapped().observedStreamTime - gracePeriod + 1;
+ }
+ }
+ }
+ }
+
+
+ @Override
+ public Maybe<ValueAndTimestamp<V>> priorValueForBuffered(final K key) {
+ return Maybe.undefined();
+ }
+
+ @Override
+ public void put(final long time, final Record<K, V> record, final ProcessorRecordContext recordContext) {
+ requireNonNull(record.value(), "value cannot be null");
+ requireNonNull(record.key(), "key cannot be null");
+ requireNonNull(recordContext, "recordContext cannot be null");
+ if (wrapped().observedStreamTime - gracePeriod > record.timestamp()) {
+ return;
+ }
+ maybeUpdateSeqnumForDups();
+ final Bytes serializedKey = Bytes.wrap(
+ PrefixedWindowKeySchemas.TimeFirstWindowKeySchema.toStoreKeyBinary(keySerde.serializer().serialize(topic, record.key()),
+ record.timestamp(),
+ seqnum).get());
+ final byte[] valueBytes = valueSerde.serializer().serialize(topic, record.value());
+ final BufferValue buffered = new BufferValue(null, null, valueBytes, recordContext);
+ wrapped().put(serializedKey, buffered.serialize(0).array());
+ bufferSize += computeRecordSize(serializedKey, buffered);
+ numRecords++;
+ if (minTimestamp() > record.timestamp()) {
+ minTimestamp = record.timestamp();
+ }
+ }
+
+ @Override
+ public int numRecords() {
+ return numRecords;
+ }
+
+ @Override
+ public long bufferSize() {
+ return bufferSize;
+ }
+
+ @Override
+ public long minTimestamp() {
+ return minTimestamp;
+ }
+
+ private static long computeRecordSize(final Bytes key, final BufferValue value) {
+ long size = 0L;
+ size += key.get().length;
+ if (value != null) {
+ size += value.residentMemorySizeEstimate();
+ }
+ return size;
+ }
+
+ private void maybeUpdateSeqnumForDups() {
+ seqnum = (seqnum + 1) & 0x7FFFFFFF;
+ }
+}
\ No newline at end of file
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java
new file mode 100644
index 00000000000..5906f8c8b9a
--- /dev/null
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java
@@ -0,0 +1,52 @@
+/*
+ * 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 java.util.Collection;
+import java.util.Map;
+import java.util.Optional;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.state.internals.PrefixedWindowKeySchemas.TimeFirstWindowKeySchema;
+import org.rocksdb.WriteBatch;
+
+/**
+ * A RocksDB backed time-ordered segmented bytes store for window key schema.
+ */
+public class RocksDBTimeOrderedKeyValueBytesStore extends AbstractRocksDBTimeOrderedSegmentedBytesStore {
+
+ RocksDBTimeOrderedKeyValueBytesStore(final String name,
+ final String metricsScope) {
+ super(name, metricsScope, Long.MAX_VALUE, Long.MAX_VALUE, new TimeFirstWindowKeySchema(), Optional.empty());
+ }
+
+ @Override
+ protected KeyValue<Bytes, byte[]> getIndexKeyValue(final Bytes baseKey, final byte[] baseValue) {
+ throw new UnsupportedOperationException("Do not use for TimeOrderedKeyValueStore");
+ }
+
+ @Override
+ Map<KeyValueSegment, WriteBatch> getWriteBatches(final Collection<ConsumerRecord<byte[], byte[]>> records) {
+ throw new UnsupportedOperationException("Do not use for TimeOrderedKeyValueStore");
+ }
+
+ @Override
+ protected IndexToBaseStoreIterator getIndexToBaseStoreIterator(final SegmentIterator<KeyValueSegment> segmentIterator) {
+ throw new UnsupportedOperationException("Do not use for TimeOrderedKeyValueStore");
+ }
+}
\ No newline at end of file
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbKeyValueBytesStoreSupplier.java b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreSupplier.java
similarity index 57%
rename from streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbKeyValueBytesStoreSupplier.java
rename to streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreSupplier.java
index 87be76781fe..3ce6e4358a6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbKeyValueBytesStoreSupplier.java
+++ b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStoreSupplier.java
@@ -16,35 +16,29 @@
*/
package org.apache.kafka.streams.state.internals;
-import org.apache.kafka.common.utils.Bytes;
-import org.apache.kafka.streams.state.KeyValueBytesStoreSupplier;
-import org.apache.kafka.streams.state.KeyValueStore;
-
-public class RocksDbKeyValueBytesStoreSupplier implements KeyValueBytesStoreSupplier {
-
+public class RocksDBTimeOrderedKeyValueBytesStoreSupplier {
private final String name;
- private final boolean returnTimestampedStore;
+ private final long retentionPeriod;
- public RocksDbKeyValueBytesStoreSupplier(final String name,
- final boolean returnTimestampedStore) {
+ public RocksDBTimeOrderedKeyValueBytesStoreSupplier(final String name,
+ final long retentionPeriod) {
this.name = name;
- this.returnTimestampedStore = returnTimestampedStore;
+ this.retentionPeriod = retentionPeriod;
}
- @Override
public String name() {
return name;
}
- @Override
- public KeyValueStore<Bytes, byte[]> get() {
- return returnTimestampedStore ?
- new RocksDBTimestampedStore(name, metricsScope()) :
- new RocksDBStore(name, metricsScope());
+ public RocksDBTimeOrderedKeyValueBytesStore get() {
+ return new RocksDBTimeOrderedKeyValueBytesStore(
+ name,
+ metricsScope()
+ );
}
- @Override
public String metricsScope() {
return "rocksdb";
}
-}
+
+}
\ No newline at end of file
diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java
new file mode 100644
index 00000000000..9c15b4c5c17
--- /dev/null
+++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java
@@ -0,0 +1,191 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.streams.state.internals;
+
+import org.apache.kafka.common.header.internals.RecordHeaders;
+import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.metrics.Sensor;
+import org.apache.kafka.common.serialization.Serdes;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.processor.TaskId;
+import org.apache.kafka.streams.processor.api.Record;
+import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
+import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
+import org.apache.kafka.streams.processor.internals.SerdeGetter;
+import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
+import org.apache.kafka.test.MockInternalNewProcessorContext;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.equalTo;
+import static org.mockito.Mockito.when;
+
+@RunWith(MockitoJUnitRunner.StrictStubs.class)
+public class RocksDBTimeOrderedKeyValueBufferTest {
+ public RocksDBTimeOrderedKeyValueBuffer<String, String> buffer;
+ @Mock
+ public SerdeGetter serdeGetter;
+ public InternalProcessorContext<String, String> context;
+ public StreamsMetricsImpl streamsMetrics;
+ @Mock
+ public Sensor sensor;
+ public long offset;
+
+ @Before
+ public void setUp() {
+ when(serdeGetter.keySerde()).thenReturn(new Serdes.StringSerde());
+ when(serdeGetter.valueSerde()).thenReturn(new Serdes.StringSerde());
+ final Metrics metrics = new Metrics();
+ offset = 0;
+ streamsMetrics = new StreamsMetricsImpl(metrics, "test-client", StreamsConfig.METRICS_LATEST, new MockTime());
+ context = new MockInternalNewProcessorContext<>(StreamsTestUtils.getStreamsConfig(), new TaskId(0, 0), TestUtils.tempDirectory());
+ }
+
+ private void createBuffer(final Duration grace) {
+ final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDBTimeOrderedKeyValueBytesStoreSupplier("testing", 100).get();
+ buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing");
+ buffer.setSerdesIfNull(serdeGetter);
+ buffer.init((StateStoreContext) context, store);
+ }
+
+ private void pipeRecord(final String key, final String value, final long time) {
+ final Record<String, String> record = new Record<>(key, value, time);
+ context.setRecordContext(new ProcessorRecordContext(time, offset++, 0, "testing", new RecordHeaders()));
+ buffer.put(time, record, context.recordContext());
+ }
+
+ @Test
+ public void shouldPutInBufferAndUpdateFields() {
+ createBuffer(Duration.ofMinutes(1));
+ assertNumSizeAndTimestamp(buffer, 0, Long.MAX_VALUE, 0);
+ pipeRecord("1", "0", 0L);
+ assertNumSizeAndTimestamp(buffer, 1, 0, 42);
+ pipeRecord("3", "0", 2L);
+ assertNumSizeAndTimestamp(buffer, 2, 0, 84);
+ }
+
+ @Test
+ public void shouldAddAndEvictRecord() {
+ createBuffer(Duration.ZERO);
+ final AtomicInteger count = new AtomicInteger(0);
+ pipeRecord("1", "0", 0L);
+ assertNumSizeAndTimestamp(buffer, 1, 0, 42);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertNumSizeAndTimestamp(buffer, 0, Long.MAX_VALUE, 0);
+ assertThat(count.get(), equalTo(1));
+ }
+
+ @Test
+ public void shouldAddAndEvictRecordTwice() {
+ createBuffer(Duration.ZERO);
+ final AtomicInteger count = new AtomicInteger(0);
+ pipeRecord("1", "0", 0L);
+ assertNumSizeAndTimestamp(buffer, 1, 0, 42);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertNumSizeAndTimestamp(buffer, 0, Long.MAX_VALUE, 0);
+ assertThat(count.get(), equalTo(1));
+ pipeRecord("2", "0", 1L);
+ assertNumSizeAndTimestamp(buffer, 1, 1, 42);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertNumSizeAndTimestamp(buffer, 0, Long.MAX_VALUE, 0);
+ assertThat(count.get(), equalTo(2));
+ }
+
+ @Test
+ public void shouldAddAndEvictRecordTwiceWithNonZeroGrace() {
+ createBuffer(Duration.ofMillis(1));
+ final AtomicInteger count = new AtomicInteger(0);
+ pipeRecord("1", "0", 0L);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertNumSizeAndTimestamp(buffer, 1, 0, 42);
+ assertThat(count.get(), equalTo(0));
+ pipeRecord("2", "0", 1L);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertNumSizeAndTimestamp(buffer, 1, 1, 42);
+ assertThat(count.get(), equalTo(1));
+ }
+
+ @Test
+ public void shouldAddRecordsTwiceAndEvictRecordsOnce() {
+ createBuffer(Duration.ZERO);
+ final AtomicInteger count = new AtomicInteger(0);
+ pipeRecord("1", "0", 0L);
+ buffer.evictWhile(() -> buffer.numRecords() > 1, r -> count.getAndIncrement());
+ assertThat(count.get(), equalTo(0));
+ pipeRecord("2", "0", 1L);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertThat(count.get(), equalTo(2));
+ }
+
+ @Test
+ public void shouldDropLateRecords() {
+ createBuffer(Duration.ZERO);
+ pipeRecord("1", "0", 1L);
+ assertNumSizeAndTimestamp(buffer, 1, 1, 42);
+ pipeRecord("2", "0", 0L);
+ assertNumSizeAndTimestamp(buffer, 1, 1, 42);
+ }
+
+ @Test
+ public void shouldDropLateRecordsWithNonZeroGrace() {
+ createBuffer(Duration.ofMillis(1));
+ pipeRecord("1", "0", 2L);
+ assertNumSizeAndTimestamp(buffer, 1, 2, 42);
+ pipeRecord("2", "0", 1L);
+ assertNumSizeAndTimestamp(buffer, 2, 1, 84);
+ pipeRecord("3", "0", 0L);
+ assertNumSizeAndTimestamp(buffer, 2, 1, 84);
+ }
+
+ @Test
+ public void shouldHandleCollidingKeys() {
+ createBuffer(Duration.ofMillis(1));
+ final AtomicInteger count = new AtomicInteger(0);
+ pipeRecord("2", "0", 0L);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertThat(count.get(), equalTo(0));
+ assertNumSizeAndTimestamp(buffer, 1, 0, 42);
+ pipeRecord("2", "2", 0L);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertThat(count.get(), equalTo(0));
+ assertNumSizeAndTimestamp(buffer, 2, 0, 84);
+ pipeRecord("1", "0", 7L);
+ assertNumSizeAndTimestamp(buffer, 3, 0, 126);
+ buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement());
+ assertThat(count.get(), equalTo(2));
+ assertNumSizeAndTimestamp(buffer, 1, 7, 42);
+ }
+
+ private void assertNumSizeAndTimestamp(final TimeOrderedKeyValueBuffer<String, String, String> buffer,
+ final int num,
+ final long time,
+ final long size) {
+ assertThat(buffer.numRecords(), equalTo(num));
+ assertThat(buffer.minTimestamp(), equalTo(time));
+ assertThat(buffer.bufferSize(), equalTo(size));
+ }
+}
\ No newline at end of file