TimeSeriesStoreImpl implements TimeSeriesStore<K, V> { @Override public void put(K key, V val, long timestamp) { if (appendMode) { seqNum.getAndIncrement(); } TimeSeriesKey<K> timeSeriesKey = new TimeSeriesKey<>(key, timestamp, seqNum.get()); LOG.trace("Inserting {} -> {} into the store", timeSeriesKey, val); kvStore.put(timeSeriesKey, val); } TimeSeriesStoreImpl(KeyValueStore<TimeSeriesKey<K>, V> kvStore, boolean appendMode);  TimeSeriesStoreImpl(KeyValueStore<TimeSeriesKey<K>, V> kvStore); @Override void put(K key, V val, long timestamp); @Override ClosableIterator<TimestampedValue<V>> get(K key, long startTimestamp, long endTimestamp); @Override ClosableIterator<TimestampedValue<V>> get(K key, long startTimestamp, long endTimestamp, int maxValues); @Override ClosableIterator<TimestampedValue<V>> get(K key, long timestamp); @Override void remove(K key, long startTimestamp, long endTimeStamp); @Override void remove(K key, long timestamp); @Override void flush(); @Override void close();  }
@Test public void testGetWithNonExistentKeys() { TimeSeriesStore<String, byte[]> timeSeriesStore = newTimeSeriesStore(new StringSerde("UTF-8"), true); timeSeriesStore.put("hello", "world-1".getBytes(), 1L); List<TimestampedValue<byte[]>> values = readStore(timeSeriesStore, "non-existent-key", 0, Integer.MAX_VALUE); Assert.assertEquals(0, values.size()); values = readStore(timeSeriesStore, "hello", 2, Integer.MAX_VALUE); Assert.assertEquals(0, values.size()); }
@Test public void testPutWithMultipleEntries() { TimeSeriesStore<String, byte[]> timeSeriesStore = newTimeSeriesStore(new StringSerde("UTF-8"), true); for (int i = 0; i < 100; i++) { timeSeriesStore.put("hello", "world-1".getBytes(), 1L); timeSeriesStore.put("hello", "world-2".getBytes(), 2L); } List<TimestampedValue<byte[]>> values = readStore(timeSeriesStore, "hello", 0L, 2L); Assert.assertEquals(100, values.size()); values.forEach(timeSeriesValue -> { Assert.assertEquals("world-1", new String(timeSeriesValue.getValue())); }); values = readStore(timeSeriesStore, "hello", 2L, 4L); Assert.assertEquals(100, values.size()); values.forEach(timeSeriesValue -> { Assert.assertEquals("world-2", new String(timeSeriesValue.getValue())); }); values = readStore(timeSeriesStore, "hello", 0L, Integer.MAX_VALUE); Assert.assertEquals(200, values.size()); }