cadonna commented on code in PR #13756: URL: https://github.com/apache/kafka/pull/13756#discussion_r1229370444
########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { + final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDbTimeOrderedKeyValueBytesStoreSupplier("testing", 100).get(); + buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing"); + buffer.setSerdesIfNull(serdeGetter); + store.init((StateStoreContext) context, store); + 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 shouldAddAndEvictRecord() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } Review Comment: Could you please add verifications that `numRecords()`, `bufferSize()`, and `minTimestamp()` are updated correctly? ########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { + final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDbTimeOrderedKeyValueBytesStoreSupplier("testing", 100).get(); + buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing"); + buffer.setSerdesIfNull(serdeGetter); + store.init((StateStoreContext) context, store); + buffer.init((StateStoreContext) context, store); Review Comment: Why do we need to initialise both -- buffer and store -- separately? Shouldn't the store be initialised when the buffer is initialised? ########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { + final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDbTimeOrderedKeyValueBytesStoreSupplier("testing", 100).get(); + buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing"); + buffer.setSerdesIfNull(serdeGetter); + store.init((StateStoreContext) context, store); + 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 shouldAddAndEvictRecord() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldAddAndEvictRecordTwice() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(2)); + } Review Comment: Could you please add verifications after the first evict AND after the second evict that `numRecords()`, `bufferSize()`, and `minTimestamp()` are updated correctly? ########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { + final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDbTimeOrderedKeyValueBytesStoreSupplier("testing", 100).get(); + buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing"); + buffer.setSerdesIfNull(serdeGetter); + store.init((StateStoreContext) context, store); + 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 shouldAddAndEvictRecord() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldAddAndEvictRecordTwice() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(2)); + } + + @Test + public void shouldAddAndEvictRecordTwiceWithNonZeroGrace() { + createJoin(Duration.ofMillis(1)); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(0)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldAddRecordsTwiceAndEvictRecordsOnce() { + createJoin(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() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 1, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(0)); + pipeRecord("2", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldDropLateRecordsWithNonZeroGrace() { + createJoin(Duration.ofMillis(1)); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 2L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(0)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + pipeRecord("2", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldHandleCollidingKeys() { Review Comment: Also here, I think it is enough to verify `numRecords()`. ########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { Review Comment: The name is a bit misleading. Could you please rename it to `createBuffer()`? ########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { + final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDbTimeOrderedKeyValueBytesStoreSupplier("testing", 100).get(); + buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing"); + buffer.setSerdesIfNull(serdeGetter); + store.init((StateStoreContext) context, store); + buffer.init((StateStoreContext) context, store); + } + Review Comment: Could you please add a test for `buffer.put()`? In that test you first verify that `numRecords()`, `bufferSize()`, and `minTimestamp()` have the intial values. Then you add a record to the buffer and test again `numRecords()`, `bufferSize()`, and `minTimestamp()` and the same for a second record. ########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { Review Comment: ```suggestion private void createJoin(final Duration grace) { ``` ########## streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBuffer.java: ########## @@ -0,0 +1,172 @@ +/* + * 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)) { Review Comment: Do we have a guarantee that `fetchAll()` will return the result record in timestamp order? I understand that it will return records in a range but will those records be ordered? The case I have in mind is when some records are in SST files and others are in the memtable. ########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { + final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDbTimeOrderedKeyValueBytesStoreSupplier("testing", 100).get(); + buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing"); + buffer.setSerdesIfNull(serdeGetter); + store.init((StateStoreContext) context, store); + 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 shouldAddAndEvictRecord() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldAddAndEvictRecordTwice() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(2)); + } + + @Test + public void shouldAddAndEvictRecordTwiceWithNonZeroGrace() { + createJoin(Duration.ofMillis(1)); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(0)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldAddRecordsTwiceAndEvictRecordsOnce() { + createJoin(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() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 1, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(0)); + pipeRecord("2", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldDropLateRecordsWithNonZeroGrace() { + createJoin(Duration.ofMillis(1)); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 2L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(0)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + pipeRecord("2", "0", 0L); Review Comment: I think you should use key `3` here. Just for clarity, I found myself thinking why you used again 2 as key here. ########## streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDbTimeOrderedKeyValueBytesStoreSupplier.java: ########## @@ -0,0 +1,46 @@ +/* + * 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; + +public class RocksDbTimeOrderedKeyValueBytesStoreSupplier { Review Comment: Why did you mark this as resolved? ########## streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBufferTest.java: ########## @@ -0,0 +1,168 @@ +/* + * 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()); + } + + public void createJoin(final Duration grace) { + final RocksDBTimeOrderedKeyValueBytesStore store = new RocksDbTimeOrderedKeyValueBytesStoreSupplier("testing", 100).get(); + buffer = new RocksDBTimeOrderedKeyValueBuffer<>(store, grace, "testing"); + buffer.setSerdesIfNull(serdeGetter); + store.init((StateStoreContext) context, store); + 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 shouldAddAndEvictRecord() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldAddAndEvictRecordTwice() { + createJoin(Duration.ZERO); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(2)); + } + + @Test + public void shouldAddAndEvictRecordTwiceWithNonZeroGrace() { + createJoin(Duration.ofMillis(1)); + final AtomicInteger count = new AtomicInteger(0); + pipeRecord("1", "0", 0L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(0)); + pipeRecord("2", "0", 1L); + buffer.evictWhile(() -> buffer.numRecords() > 0, r -> count.getAndIncrement()); + assertThat(count.get(), equalTo(1)); + } + + @Test + public void shouldAddRecordsTwiceAndEvictRecordsOnce() { + createJoin(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() { Review Comment: I think you do not need to use `evictWhile()` for this test. You can just verify if `numRecords()` does not increase after calling `put()`. The same is true for the next test. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
