### Question:
MVCCAsyncBytesStoreImpl extends AbstractStateStoreWithJournal<MVCCStoreImpl<byte[], byte[]>> implements MVCCAsyncStore<byte[], byte[]> { @Override public CompletableFuture<RangeResult<byte[], byte[]>> range(RangeOp<byte[], byte[]> rangeOp) { synchronized (this) { if (!isInitialized) { return FutureUtils.exception(new InvalidStateStoreException("State store is not initialized yet.")); } } return executeReadIO(() -> localStore.range(rangeOp)); } MVCCAsyncBytesStoreImpl(Supplier<MVCCStoreImpl<byte[], byte[]>> storeSupplier,
                            Supplier<Namespace> namespaceSupplier); @VisibleForTesting @Override boolean ownWriteScheduler(); @VisibleForTesting @Override boolean ownReadScheduler(); @Override OpFactory<byte[], byte[]> getOpFactory(); @Override CompletableFuture<RangeResult<byte[], byte[]>> range(RangeOp<byte[], byte[]> rangeOp); @Override CompletableFuture<PutResult<byte[], byte[]>> put(PutOp<byte[], byte[]> op); @Override CompletableFuture<DeleteResult<byte[], byte[]>> delete(DeleteOp<byte[], byte[]> op); @Override CompletableFuture<TxnResult<byte[], byte[]>> txn(TxnOp<byte[], byte[]> op); @Override CompletableFuture<IncrementResult<byte[], byte[]>> increment(IncrementOp<byte[], byte[]> op);  }### Answer:
@Test public void testPutGetDeleteRanges() throws Exception { this.streamName = "test-put-kvs"; StateStoreSpec spec = initSpec(streamName); result(store.init(spec)); int numPairs = 100; List<PutResult<byte[], byte[]>> kvs = writeKVs(numPairs, true); assertEquals(numPairs, kvs.size()); for (PutResult<byte[], byte[]> kv : kvs) { assertEquals(Code.OK, kv.code()); assertNull(kv.prevKv()); kv.close(); } verifyRange(20, 70, 2, 2, 0); List<KeyValue<byte[], byte[]>> prevKvs = result(store.deleteRange(getKey(20), getKey(70))); assertNotNull(prevKvs); verifyRecords( prevKvs, 20, 70, 2, 2, 0); prevKvs.forEach(KeyValue::close); prevKvs = result(store.range(getKey(20), getKey(70))); assertTrue(prevKvs.isEmpty()); }
@Test public void testReplayJournal() throws Exception { this.streamName = "test-replay-journal"; StateStoreSpec spec = initSpec(streamName); result(store.init(spec)); int numKvs = 10; IntStream.range(0, numKvs) .forEach(i -> { try { result(store.putIfAbsent(getKey(i), getValue(100 + i))); } catch (Exception e) { log.error("Failed to put kv pair ({})", i, e); } }); log.info("Closing the store '{}' ...", streamName); store.close(); log.info("Closed the store '{}' ...", streamName); store = new MVCCAsyncBytesStoreImpl( () -> new MVCCStoreImpl<>(), () -> namespace); spec = StateStoreSpec.builder() .name(streamName) .keyCoder(ByteArrayCoder.of()) .valCoder(ByteArrayCoder.of()) .stream(streamName) .localStateStoreDir(testDir.newFolder()) .build(); result(store.init(spec)); for (int i = 0; i < numKvs; i++) { byte[] value = result(store.get(getKey(i))); assertNotNull(value); assertArrayEquals(getValue(100 + i), value); } }