@Override
    public CompletableFuture<Void> process() {
        // TODO: Use MultiGet to optimize this implement

        CompletableFuture<Void> future = new CompletableFuture<>();

        AtomicReference<Exception> error = new AtomicReference<>();
        AtomicInteger counter = new AtomicInteger(batchRequest.size());
        for (int i = 0; i < batchRequest.size(); i++) {
            ForStDBGetRequest<?, ?, ?> request = batchRequest.get(i);
            executor.execute(
                    () -> {
                        try {
                            if (error.get() == null) {
                                byte[] key = request.buildSerializedKey();
                                byte[] value = db.get(request.getColumnFamilyHandle(), key);
                                request.completeStateFuture(value);
                            } else {
                                request.completeStateFutureExceptionally(
                                        "Error already occurred in other state request of the same "
                                                + "group, failed the state request directly",
                                        error.get());
                            }
                        } catch (Exception e) {
                            error.set(e);
                            request.completeStateFutureExceptionally(
                                    "Error when execute ForStDb get operation", e);
                            future.completeExceptionally(e);
                        } finally {
                            if (counter.decrementAndGet() == 0
                                    && !future.isCompletedExceptionally()) {
                                future.complete(null);
                            }
                        }
                    });
        }
        return future;
    }
@Test
    public void testValueStateMultiGet() throws Exception {
        ForStValueState<Integer, VoidNamespace, String> valueState1 =
                buildForStValueState("test-multiGet-1");
        ForStValueState<Integer, VoidNamespace, String> valueState2 =
                buildForStValueState("test-multiGet-2");
        List<ForStDBGetRequest<?, ?, ?>> batchGetRequest = new ArrayList<>();
        List<Tuple2<String, TestStateFuture<String>>> resultCheckList = new ArrayList<>();

        int keyNum = 1000;
        for (int i = 0; i < keyNum; i++) {
            TestStateFuture<String> future = new TestStateFuture<>();
            ForStValueState<Integer, VoidNamespace, String> table =
                    ((i % 2 == 0) ? valueState1 : valueState2);
            ForStDBGetRequest<Integer, VoidNamespace, String> request =
                    ForStDBGetRequest.of(buildContextKey(i), table, future);
            batchGetRequest.add(request);

            String value = (i % 10 != 0 ? String.valueOf(i) : null);
            resultCheckList.add(Tuple2.of(value, future));
            if (value == null) {
                continue;
            }
            byte[] keyBytes = request.buildSerializedKey();
            byte[] valueBytes = table.serializeValue(value);
            db.put(request.getColumnFamilyHandle(), keyBytes, valueBytes);
        }

        ExecutorService executor = Executors.newFixedThreadPool(4);
        ForStGeneralMultiGetOperation generalMultiGetOperation =
                new ForStGeneralMultiGetOperation(db, batchGetRequest, executor);
        generalMultiGetOperation.process().get();

        for (Tuple2<String, TestStateFuture<String>> tuple : resultCheckList) {
            assertThat(tuple.f1.getCompletedResult()).isEqualTo(tuple.f0);
        }

        executor.shutdownNow();
    }