| /** |
| * 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.clients.producer.MockProducer; |
| import org.apache.kafka.clients.producer.Producer; |
| import org.apache.kafka.clients.producer.ProducerRecord; |
| import org.apache.kafka.common.serialization.Serde; |
| import org.apache.kafka.common.serialization.Serdes; |
| import org.apache.kafka.common.serialization.Serializer; |
| import org.apache.kafka.common.utils.Utils; |
| import org.apache.kafka.streams.KeyValue; |
| import org.apache.kafka.streams.processor.ProcessorContext; |
| import org.apache.kafka.streams.processor.StateStoreSupplier; |
| import org.apache.kafka.streams.processor.internals.RecordCollector; |
| import org.apache.kafka.streams.state.StateSerdes; |
| import org.apache.kafka.streams.state.WindowStore; |
| import org.apache.kafka.streams.state.WindowStoreIterator; |
| import org.apache.kafka.streams.state.WindowStoreUtils; |
| import org.apache.kafka.test.MockProcessorContext; |
| import org.junit.Test; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.nio.file.Files; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| |
| import static org.junit.Assert.assertEquals; |
| import static org.junit.Assert.assertNull; |
| |
| public class RocksDBWindowStoreTest { |
| |
| private final Serde<byte[]> byteArraySerde = Serdes.ByteArray(); |
| private final String windowName = "window"; |
| private final int numSegments = 3; |
| private final long segmentSize = RocksDBWindowStore.MIN_SEGMENT_INTERVAL; |
| private final long retentionPeriod = segmentSize * (numSegments - 1); |
| private final long windowSize = 3; |
| private final Serde<Integer> intSerde = Serdes.Integer(); |
| private final Serde<String> stringSerde = Serdes.String(); |
| private final StateSerdes<Integer, String> serdes = new StateSerdes<>("", intSerde, stringSerde); |
| |
| @SuppressWarnings("unchecked") |
| protected <K, V> WindowStore<K, V> createWindowStore(ProcessorContext context) { |
| StateStoreSupplier supplier = new RocksDBWindowStoreSupplier<>(windowName, retentionPeriod, numSegments, true, intSerde, stringSerde); |
| |
| WindowStore<K, V> store = (WindowStore<K, V>) supplier.get(); |
| store.init(context, store); |
| return store; |
| } |
| |
| @Test |
| public void testPutAndFetch() throws IOException { |
| File baseDir = Files.createTempDirectory("test").toFile(); |
| try { |
| final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>(); |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| changeLog.add(new KeyValue<>( |
| keySerializer.serialize(record.topic(), record.key()), |
| valueSerializer.serialize(record.topic(), record.value())) |
| ); |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| try { |
| long startTime = segmentSize - 4L; |
| |
| context.setTime(startTime + 0L); |
| store.put(0, "zero"); |
| context.setTime(startTime + 1L); |
| store.put(1, "one"); |
| context.setTime(startTime + 2L); |
| store.put(2, "two"); |
| context.setTime(startTime + 3L); |
| // (3, "three") is not put |
| context.setTime(startTime + 4L); |
| store.put(4, "four"); |
| context.setTime(startTime + 5L); |
| store.put(5, "five"); |
| |
| assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L - windowSize, startTime + 0L + windowSize))); |
| assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L - windowSize, startTime + 1L + windowSize))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L - windowSize, startTime + 3L + windowSize))); |
| assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L - windowSize, startTime + 4L + windowSize))); |
| assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L - windowSize, startTime + 5L + windowSize))); |
| |
| context.setTime(startTime + 3L); |
| store.put(2, "two+1"); |
| context.setTime(startTime + 4L); |
| store.put(2, "two+2"); |
| context.setTime(startTime + 5L); |
| store.put(2, "two+3"); |
| context.setTime(startTime + 6L); |
| store.put(2, "two+4"); |
| context.setTime(startTime + 7L); |
| store.put(2, "two+5"); |
| context.setTime(startTime + 8L); |
| store.put(2, "two+6"); |
| |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 2L - windowSize, startTime - 2L + windowSize))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime - 1L - windowSize, startTime - 1L + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime - windowSize, startTime + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 1L - windowSize, startTime + 1L + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 3L - windowSize, startTime + 3L + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 4L - windowSize, startTime + 4L + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 5L - windowSize, startTime + 5L + windowSize))); |
| assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 6L - windowSize, startTime + 6L + windowSize))); |
| assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 7L - windowSize, startTime + 7L + windowSize))); |
| assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 8L - windowSize, startTime + 8L + windowSize))); |
| assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 9L - windowSize, startTime + 9L + windowSize))); |
| assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 10L - windowSize, startTime + 10L + windowSize))); |
| assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 11L - windowSize, startTime + 11L + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L - windowSize, startTime + 12L + windowSize))); |
| |
| // Flush the store and verify all current entries were properly flushed ... |
| store.flush(); |
| |
| Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime); |
| |
| assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0)); |
| assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1)); |
| assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2)); |
| assertNull(entriesByKey.get(3)); |
| assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4)); |
| assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5)); |
| assertNull(entriesByKey.get(6)); |
| |
| } finally { |
| store.close(); |
| } |
| |
| } finally { |
| Utils.delete(baseDir); |
| } |
| } |
| |
| @Test |
| public void testPutAndFetchBefore() throws IOException { |
| File baseDir = Files.createTempDirectory("test").toFile(); |
| try { |
| final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>(); |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| changeLog.add(new KeyValue<>( |
| keySerializer.serialize(record.topic(), record.key()), |
| valueSerializer.serialize(record.topic(), record.value())) |
| ); |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| try { |
| long startTime = segmentSize - 4L; |
| |
| context.setTime(startTime + 0L); |
| store.put(0, "zero"); |
| context.setTime(startTime + 1L); |
| store.put(1, "one"); |
| context.setTime(startTime + 2L); |
| store.put(2, "two"); |
| context.setTime(startTime + 3L); |
| // (3, "three") is not put |
| context.setTime(startTime + 4L); |
| store.put(4, "four"); |
| context.setTime(startTime + 5L); |
| store.put(5, "five"); |
| |
| assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L - windowSize, startTime + 0L))); |
| assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L - windowSize, startTime + 1L))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L))); |
| assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L - windowSize, startTime + 3L))); |
| assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L - windowSize, startTime + 4L))); |
| assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L - windowSize, startTime + 5L))); |
| |
| context.setTime(startTime + 3L); |
| store.put(2, "two+1"); |
| context.setTime(startTime + 4L); |
| store.put(2, "two+2"); |
| context.setTime(startTime + 5L); |
| store.put(2, "two+3"); |
| context.setTime(startTime + 6L); |
| store.put(2, "two+4"); |
| context.setTime(startTime + 7L); |
| store.put(2, "two+5"); |
| context.setTime(startTime + 8L); |
| store.put(2, "two+6"); |
| |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 1L - windowSize, startTime - 1L))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 0L - windowSize, startTime + 0L))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 1L - windowSize, startTime + 1L))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L - windowSize, startTime + 2L))); |
| assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime + 3L - windowSize, startTime + 3L))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 4L - windowSize, startTime + 4L))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 5L - windowSize, startTime + 5L))); |
| assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 6L - windowSize, startTime + 6L))); |
| assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 7L - windowSize, startTime + 7L))); |
| assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 8L - windowSize, startTime + 8L))); |
| assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 9L - windowSize, startTime + 9L))); |
| assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 10L - windowSize, startTime + 10L))); |
| assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 11L - windowSize, startTime + 11L))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L - windowSize, startTime + 12L))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 13L - windowSize, startTime + 13L))); |
| |
| // Flush the store and verify all current entries were properly flushed ... |
| store.flush(); |
| |
| Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime); |
| |
| assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0)); |
| assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1)); |
| assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2)); |
| assertNull(entriesByKey.get(3)); |
| assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4)); |
| assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5)); |
| assertNull(entriesByKey.get(6)); |
| |
| } finally { |
| store.close(); |
| } |
| |
| } finally { |
| Utils.delete(baseDir); |
| } |
| } |
| |
| @Test |
| public void testPutAndFetchAfter() throws IOException { |
| File baseDir = Files.createTempDirectory("test").toFile(); |
| try { |
| final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>(); |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| changeLog.add(new KeyValue<>( |
| keySerializer.serialize(record.topic(), record.key()), |
| valueSerializer.serialize(record.topic(), record.value())) |
| ); |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| try { |
| long startTime = segmentSize - 4L; |
| |
| context.setTime(startTime + 0L); |
| store.put(0, "zero"); |
| context.setTime(startTime + 1L); |
| store.put(1, "one"); |
| context.setTime(startTime + 2L); |
| store.put(2, "two"); |
| context.setTime(startTime + 3L); |
| // (3, "three") is not put |
| context.setTime(startTime + 4L); |
| store.put(4, "four"); |
| context.setTime(startTime + 5L); |
| store.put(5, "five"); |
| |
| assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime + 0L, startTime + 0L + windowSize))); |
| assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + 1L, startTime + 1L + windowSize))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + 2L, startTime + 2L + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + 3L, startTime + 3L + windowSize))); |
| assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + 4L, startTime + 4L + windowSize))); |
| assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + 5L, startTime + 5L + windowSize))); |
| |
| context.setTime(startTime + 3L); |
| store.put(2, "two+1"); |
| context.setTime(startTime + 4L); |
| store.put(2, "two+2"); |
| context.setTime(startTime + 5L); |
| store.put(2, "two+3"); |
| context.setTime(startTime + 6L); |
| store.put(2, "two+4"); |
| context.setTime(startTime + 7L); |
| store.put(2, "two+5"); |
| context.setTime(startTime + 8L); |
| store.put(2, "two+6"); |
| |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime - 2L, startTime - 2L + windowSize))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime - 1L, startTime - 1L + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1"), toList(store.fetch(2, startTime, startTime + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2"), toList(store.fetch(2, startTime + 1L, startTime + 1L + windowSize))); |
| assertEquals(Utils.mkList("two", "two+1", "two+2", "two+3"), toList(store.fetch(2, startTime + 2L, startTime + 2L + windowSize))); |
| assertEquals(Utils.mkList("two+1", "two+2", "two+3", "two+4"), toList(store.fetch(2, startTime + 3L, startTime + 3L + windowSize))); |
| assertEquals(Utils.mkList("two+2", "two+3", "two+4", "two+5"), toList(store.fetch(2, startTime + 4L, startTime + 4L + windowSize))); |
| assertEquals(Utils.mkList("two+3", "two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 5L, startTime + 5L + windowSize))); |
| assertEquals(Utils.mkList("two+4", "two+5", "two+6"), toList(store.fetch(2, startTime + 6L, startTime + 6L + windowSize))); |
| assertEquals(Utils.mkList("two+5", "two+6"), toList(store.fetch(2, startTime + 7L, startTime + 7L + windowSize))); |
| assertEquals(Utils.mkList("two+6"), toList(store.fetch(2, startTime + 8L, startTime + 8L + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 9L, startTime + 9L + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 10L, startTime + 10L + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 11L, startTime + 11L + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + 12L, startTime + 12L + windowSize))); |
| |
| // Flush the store and verify all current entries were properly flushed ... |
| store.flush(); |
| |
| Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime); |
| |
| assertEquals(Utils.mkSet("zero@0"), entriesByKey.get(0)); |
| assertEquals(Utils.mkSet("one@1"), entriesByKey.get(1)); |
| assertEquals(Utils.mkSet("two@2", "two+1@3", "two+2@4", "two+3@5", "two+4@6", "two+5@7", "two+6@8"), entriesByKey.get(2)); |
| assertNull(entriesByKey.get(3)); |
| assertEquals(Utils.mkSet("four@4"), entriesByKey.get(4)); |
| assertEquals(Utils.mkSet("five@5"), entriesByKey.get(5)); |
| assertNull(entriesByKey.get(6)); |
| |
| } finally { |
| store.close(); |
| } |
| |
| } finally { |
| Utils.delete(baseDir); |
| } |
| } |
| |
| @Test |
| public void testPutSameKeyTimestamp() throws IOException { |
| File baseDir = Files.createTempDirectory("test").toFile(); |
| try { |
| final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>(); |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| changeLog.add(new KeyValue<>( |
| keySerializer.serialize(record.topic(), record.key()), |
| valueSerializer.serialize(record.topic(), record.value())) |
| ); |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| try { |
| long startTime = segmentSize - 4L; |
| |
| context.setTime(startTime); |
| store.put(0, "zero"); |
| |
| assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize))); |
| |
| context.setTime(startTime); |
| store.put(0, "zero"); |
| context.setTime(startTime); |
| store.put(0, "zero+"); |
| context.setTime(startTime); |
| store.put(0, "zero++"); |
| |
| assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize))); |
| assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 1L - windowSize, startTime + 1L + windowSize))); |
| assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 2L - windowSize, startTime + 2L + windowSize))); |
| assertEquals(Utils.mkList("zero", "zero", "zero+", "zero++"), toList(store.fetch(0, startTime + 3L - windowSize, startTime + 3L + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(0, startTime + 4L - windowSize, startTime + 4L + windowSize))); |
| |
| // Flush the store and verify all current entries were properly flushed ... |
| store.flush(); |
| |
| Map<Integer, Set<String>> entriesByKey = entriesByKey(changeLog, startTime); |
| |
| assertEquals(Utils.mkSet("zero@0", "zero@0", "zero+@0", "zero++@0"), entriesByKey.get(0)); |
| |
| } finally { |
| store.close(); |
| } |
| |
| } finally { |
| Utils.delete(baseDir); |
| } |
| } |
| |
| @Test |
| public void testRolling() throws IOException { |
| File baseDir = Files.createTempDirectory("test").toFile(); |
| try { |
| final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>(); |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| changeLog.add(new KeyValue<>( |
| keySerializer.serialize(record.topic(), record.key()), |
| valueSerializer.serialize(record.topic(), record.value())) |
| ); |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| RocksDBWindowStore<Integer, String> inner = |
| (RocksDBWindowStore<Integer, String>) ((MeteredWindowStore<Integer, String>) store).inner(); |
| try { |
| long startTime = segmentSize * 2; |
| long incr = segmentSize / 2; |
| |
| context.setTime(startTime); |
| store.put(0, "zero"); |
| assertEquals(Utils.mkSet(2L), inner.segmentIds()); |
| |
| context.setTime(startTime + incr); |
| store.put(1, "one"); |
| assertEquals(Utils.mkSet(2L), inner.segmentIds()); |
| |
| context.setTime(startTime + incr * 2); |
| store.put(2, "two"); |
| assertEquals(Utils.mkSet(2L, 3L), inner.segmentIds()); |
| |
| context.setTime(startTime + incr * 3); |
| // (3, "three") is not put |
| assertEquals(Utils.mkSet(2L, 3L), inner.segmentIds()); |
| |
| context.setTime(startTime + incr * 4); |
| store.put(4, "four"); |
| assertEquals(Utils.mkSet(2L, 3L, 4L), inner.segmentIds()); |
| |
| context.setTime(startTime + incr * 5); |
| store.put(5, "five"); |
| assertEquals(Utils.mkSet(2L, 3L, 4L), inner.segmentIds()); |
| |
| assertEquals(Utils.mkList("zero"), toList(store.fetch(0, startTime - windowSize, startTime + windowSize))); |
| assertEquals(Utils.mkList("one"), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize))); |
| assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize))); |
| assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize))); |
| |
| context.setTime(startTime + incr * 6); |
| store.put(6, "six"); |
| assertEquals(Utils.mkSet(3L, 4L, 5L), inner.segmentIds()); |
| |
| assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize))); |
| assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize))); |
| assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize))); |
| assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize))); |
| |
| |
| context.setTime(startTime + incr * 7); |
| store.put(7, "seven"); |
| assertEquals(Utils.mkSet(3L, 4L, 5L), inner.segmentIds()); |
| |
| assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize))); |
| assertEquals(Utils.mkList("two"), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize))); |
| assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize))); |
| assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize))); |
| assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize))); |
| assertEquals(Utils.mkList("seven"), toList(store.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize))); |
| |
| context.setTime(startTime + incr * 8); |
| store.put(8, "eight"); |
| assertEquals(Utils.mkSet(4L, 5L, 6L), inner.segmentIds()); |
| |
| assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize))); |
| assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize))); |
| assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize))); |
| assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize))); |
| assertEquals(Utils.mkList("seven"), toList(store.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize))); |
| assertEquals(Utils.mkList("eight"), toList(store.fetch(8, startTime + incr * 8 - windowSize, startTime + incr * 8 + windowSize))); |
| |
| // check segment directories |
| store.flush(); |
| assertEquals( |
| Utils.mkSet(inner.segmentName(4L), inner.segmentName(5L), inner.segmentName(6L)), |
| segmentDirs(baseDir) |
| ); |
| } finally { |
| store.close(); |
| } |
| |
| } finally { |
| Utils.delete(baseDir); |
| } |
| } |
| |
| @Test |
| public void testRestore() throws IOException { |
| final List<KeyValue<byte[], byte[]>> changeLog = new ArrayList<>(); |
| long startTime = segmentSize * 2; |
| long incr = segmentSize / 2; |
| |
| File baseDir = Files.createTempDirectory("test").toFile(); |
| try { |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| changeLog.add(new KeyValue<>( |
| keySerializer.serialize(record.topic(), record.key()), |
| valueSerializer.serialize(record.topic(), record.value())) |
| ); |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| try { |
| context.setTime(startTime); |
| store.put(0, "zero"); |
| context.setTime(startTime + incr); |
| store.put(1, "one"); |
| context.setTime(startTime + incr * 2); |
| store.put(2, "two"); |
| context.setTime(startTime + incr * 3); |
| store.put(3, "three"); |
| context.setTime(startTime + incr * 4); |
| store.put(4, "four"); |
| context.setTime(startTime + incr * 5); |
| store.put(5, "five"); |
| context.setTime(startTime + incr * 6); |
| store.put(6, "six"); |
| context.setTime(startTime + incr * 7); |
| store.put(7, "seven"); |
| context.setTime(startTime + incr * 8); |
| store.put(8, "eight"); |
| store.flush(); |
| |
| } finally { |
| store.close(); |
| } |
| |
| |
| } finally { |
| Utils.delete(baseDir); |
| } |
| |
| File baseDir2 = Files.createTempDirectory("test").toFile(); |
| try { |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| changeLog.add(new KeyValue<>( |
| keySerializer.serialize(record.topic(), record.key()), |
| valueSerializer.serialize(record.topic(), record.value())) |
| ); |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| RocksDBWindowStore<Integer, String> inner = |
| (RocksDBWindowStore<Integer, String>) ((MeteredWindowStore<Integer, String>) store).inner(); |
| |
| try { |
| context.restore(windowName, changeLog); |
| |
| assertEquals(Utils.mkSet(4L, 5L, 6L), inner.segmentIds()); |
| |
| assertEquals(Utils.mkList(), toList(store.fetch(0, startTime - windowSize, startTime + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(1, startTime + incr - windowSize, startTime + incr + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(2, startTime + incr * 2 - windowSize, startTime + incr * 2 + windowSize))); |
| assertEquals(Utils.mkList(), toList(store.fetch(3, startTime + incr * 3 - windowSize, startTime + incr * 3 + windowSize))); |
| assertEquals(Utils.mkList("four"), toList(store.fetch(4, startTime + incr * 4 - windowSize, startTime + incr * 4 + windowSize))); |
| assertEquals(Utils.mkList("five"), toList(store.fetch(5, startTime + incr * 5 - windowSize, startTime + incr * 5 + windowSize))); |
| assertEquals(Utils.mkList("six"), toList(store.fetch(6, startTime + incr * 6 - windowSize, startTime + incr * 6 + windowSize))); |
| assertEquals(Utils.mkList("seven"), toList(store.fetch(7, startTime + incr * 7 - windowSize, startTime + incr * 7 + windowSize))); |
| assertEquals(Utils.mkList("eight"), toList(store.fetch(8, startTime + incr * 8 - windowSize, startTime + incr * 8 + windowSize))); |
| |
| // check segment directories |
| store.flush(); |
| assertEquals( |
| Utils.mkSet(inner.segmentName(4L), inner.segmentName(5L), inner.segmentName(6L)), |
| segmentDirs(baseDir) |
| ); |
| } finally { |
| store.close(); |
| } |
| |
| |
| } finally { |
| Utils.delete(baseDir2); |
| } |
| } |
| |
| @Test |
| public void testSegmentMaintenance() throws IOException { |
| File baseDir = Files.createTempDirectory("test").toFile(); |
| try { |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| // do nothing |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| RocksDBWindowStore<Integer, String> inner = |
| (RocksDBWindowStore<Integer, String>) ((MeteredWindowStore<Integer, String>) store).inner(); |
| |
| try { |
| |
| context.setTime(0L); |
| store.put(0, "v"); |
| assertEquals( |
| Utils.mkSet(inner.segmentName(0L)), |
| segmentDirs(baseDir) |
| ); |
| |
| context.setTime(59999L); |
| store.put(0, "v"); |
| context.setTime(59999L); |
| store.put(0, "v"); |
| assertEquals( |
| Utils.mkSet(inner.segmentName(0L)), |
| segmentDirs(baseDir) |
| ); |
| |
| context.setTime(60000L); |
| store.put(0, "v"); |
| assertEquals( |
| Utils.mkSet(inner.segmentName(0L), inner.segmentName(1L)), |
| segmentDirs(baseDir) |
| ); |
| |
| WindowStoreIterator iter; |
| int fetchedCount; |
| |
| iter = store.fetch(0, 0L, 240000L); |
| fetchedCount = 0; |
| while (iter.hasNext()) { |
| iter.next(); |
| fetchedCount++; |
| } |
| assertEquals(4, fetchedCount); |
| |
| assertEquals( |
| Utils.mkSet(inner.segmentName(0L), inner.segmentName(1L)), |
| segmentDirs(baseDir) |
| ); |
| |
| context.setTime(180000L); |
| store.put(0, "v"); |
| |
| iter = store.fetch(0, 0L, 240000L); |
| fetchedCount = 0; |
| while (iter.hasNext()) { |
| iter.next(); |
| fetchedCount++; |
| } |
| assertEquals(2, fetchedCount); |
| |
| assertEquals( |
| Utils.mkSet(inner.segmentName(1L), inner.segmentName(2L), inner.segmentName(3L)), |
| segmentDirs(baseDir) |
| ); |
| |
| context.setTime(300000L); |
| store.put(0, "v"); |
| |
| iter = store.fetch(0, 240000L, 1000000L); |
| fetchedCount = 0; |
| while (iter.hasNext()) { |
| iter.next(); |
| fetchedCount++; |
| } |
| assertEquals(1, fetchedCount); |
| |
| assertEquals( |
| Utils.mkSet(inner.segmentName(3L), inner.segmentName(4L), inner.segmentName(5L)), |
| segmentDirs(baseDir) |
| ); |
| |
| } finally { |
| store.close(); |
| } |
| |
| } finally { |
| Utils.delete(baseDir); |
| } |
| } |
| |
| @Test |
| public void testInitialLoading() throws IOException { |
| File baseDir = Files.createTempDirectory("test").toFile(); |
| try { |
| Producer<byte[], byte[]> producer = new MockProducer<>(true, byteArraySerde.serializer(), byteArraySerde.serializer()); |
| RecordCollector recordCollector = new RecordCollector(producer) { |
| @Override |
| public <K1, V1> void send(ProducerRecord<K1, V1> record, Serializer<K1> keySerializer, Serializer<V1> valueSerializer) { |
| // do nothing |
| } |
| }; |
| |
| MockProcessorContext context = new MockProcessorContext( |
| null, baseDir, |
| byteArraySerde, byteArraySerde, |
| recordCollector); |
| |
| File storeDir = new File(baseDir, windowName); |
| |
| WindowStore<Integer, String> store = createWindowStore(context); |
| RocksDBWindowStore<Integer, String> inner = |
| (RocksDBWindowStore<Integer, String>) ((MeteredWindowStore<Integer, String>) store).inner(); |
| |
| try { |
| new File(storeDir, inner.segmentName(0L)).mkdir(); |
| new File(storeDir, inner.segmentName(1L)).mkdir(); |
| new File(storeDir, inner.segmentName(2L)).mkdir(); |
| new File(storeDir, inner.segmentName(3L)).mkdir(); |
| new File(storeDir, inner.segmentName(4L)).mkdir(); |
| new File(storeDir, inner.segmentName(5L)).mkdir(); |
| new File(storeDir, inner.segmentName(6L)).mkdir(); |
| } finally { |
| store.close(); |
| } |
| |
| store = createWindowStore(context); |
| inner = (RocksDBWindowStore<Integer, String>) ((MeteredWindowStore<Integer, String>) store).inner(); |
| |
| try { |
| assertEquals( |
| Utils.mkSet(inner.segmentName(4L), inner.segmentName(5L), inner.segmentName(6L)), |
| segmentDirs(baseDir) |
| ); |
| |
| WindowStoreIterator iter = store.fetch(0, 0L, 1000000L); |
| while (iter.hasNext()) { |
| iter.next(); |
| } |
| |
| assertEquals( |
| Utils.mkSet(inner.segmentName(4L), inner.segmentName(5L), inner.segmentName(6L)), |
| segmentDirs(baseDir) |
| ); |
| |
| } finally { |
| store.close(); |
| } |
| |
| } finally { |
| Utils.delete(baseDir); |
| } |
| } |
| |
| private <E> List<E> toList(WindowStoreIterator<E> iterator) { |
| ArrayList<E> list = new ArrayList<>(); |
| while (iterator.hasNext()) { |
| list.add(iterator.next().value); |
| } |
| return list; |
| } |
| |
| private Set<String> segmentDirs(File baseDir) { |
| File windowDir = new File(baseDir, windowName); |
| |
| return new HashSet<>(Arrays.asList(windowDir.list())); |
| } |
| |
| private Map<Integer, Set<String>> entriesByKey(List<KeyValue<byte[], byte[]>> changeLog, long startTime) { |
| HashMap<Integer, Set<String>> entriesByKey = new HashMap<>(); |
| |
| for (KeyValue<byte[], byte[]> entry : changeLog) { |
| long timestamp = WindowStoreUtils.timestampFromBinaryKey(entry.key); |
| Integer key = WindowStoreUtils.keyFromBinaryKey(entry.key, serdes); |
| String value = entry.value == null ? null : serdes.valueFrom(entry.value); |
| |
| Set<String> entries = entriesByKey.get(key); |
| if (entries == null) { |
| entries = new HashSet<>(); |
| entriesByKey.put(key, entries); |
| } |
| entries.add(value + "@" + (timestamp - startTime)); |
| } |
| |
| return entriesByKey; |
| } |
| } |
