@Override
    public CompletableFuture<Void> process() {
        return CompletableFuture.runAsync(
                () -> {
                    try (WriteBatch writeBatch =
                            new WriteBatch(batchRequest.size() * PER_RECORD_ESTIMATE_BYTES)) {
                        for (ForStDBPutRequest<?, ?, ?> request : batchRequest) {
                            if (request.valueIsNull()) {
                                // put(key, null) == delete(key)
                                writeBatch.delete(
                                        request.getColumnFamilyHandle(),
                                        request.buildSerializedKey());
                            } else {
                                writeBatch.put(
                                        request.getColumnFamilyHandle(),
                                        request.buildSerializedKey(),
                                        request.buildSerializedValue());
                            }
                        }
                        db.write(writeOptions, writeBatch);
                        for (ForStDBPutRequest<?, ?, ?> request : batchRequest) {
                            request.completeStateFuture();
                        }
                    } catch (Exception e) {
                        String msg = "Error while write batch data to ForStDB.";
                        for (ForStDBPutRequest<?, ?, ?> request : batchRequest) {
                            // fail every state request in this batch
                            request.completeStateFutureExceptionally(msg, e);
                        }
                        // fail the whole batch operation
                        throw new CompletionException(msg, e);
                    }
                },
                executor);
    }
@Test
    public void testValueStateWriteBatch() throws Exception {
        ForStValueState<Integer, VoidNamespace, String> valueState1 =
                buildForStValueState("test-write-batch-1");
        ForStValueState<Integer, VoidNamespace, String> valueState2 =
                buildForStValueState("test-write-batch-2");
        List<ForStDBPutRequest<?, ?, ?>> batchPutRequest = new ArrayList<>();
        int keyNum = 100;
        for (int i = 0; i < keyNum; i++) {
            batchPutRequest.add(
                    ForStDBPutRequest.of(
                            buildContextKey(i),
                            String.valueOf(i),
                            ((i % 2 == 0) ? valueState1 : valueState2),
                            new TestStateFuture<>()));
        }

        ExecutorService executor = Executors.newFixedThreadPool(2);
        ForStWriteBatchOperation writeBatchOperation =
                new ForStWriteBatchOperation(db, batchPutRequest, new WriteOptions(), executor);
        writeBatchOperation.process().get();

        // check data correctness
        for (ForStDBPutRequest<?, ?, ?> request : batchPutRequest) {
            byte[] keyBytes = request.buildSerializedKey();
            byte[] valueBytes = db.get(request.getColumnFamilyHandle(), keyBytes);
            assertArrayEquals(valueBytes, request.buildSerializedValue());
        }
    }