public void asyncAddData(T data, AddDataCallback callback, Object ctx){
        if (!batchEnabled){
            if (state == State.CLOSING || state == State.CLOSED){
                callback.addFailed(BUFFERED_WRITER_CLOSED_EXCEPTION, ctx);
                return;
            }
            ByteBuf byteBuf = dataSerializer.serialize(data);
            managedLedger.asyncAddEntry(byteBuf, DisabledBatchCallback.INSTANCE,
                    AsyncAddArgs.newInstance(callback, ctx, System.currentTimeMillis(), byteBuf));
            return;
        }
        CompletableFuture
                .runAsync(
                        () -> internalAsyncAddData(data, callback, ctx), singleThreadExecutorForWrite)
                .exceptionally(e -> {
                    log.warn("Execute 'internalAsyncAddData' fail", e);
                    return null;
                });
    }
@Test
    public void testFailWhenAddData() throws Exception {
        int batchedWriteMaxSize = 1024;
        TxnLogBufferedWriter.DataSerializer dataSerializer =
                new WrongDataSerializer(batchedWriteMaxSize, true, true, true);
        int writeCount = 100;
        var callbackWithCounter = createCallBackWithCounter();
        // Create TxnLogBufferedWriter.
        var txnLogBufferedWriterContext = createTxnBufferedWriterContextWithMetrics(
                dataSerializer, Integer.MAX_VALUE, batchedWriteMaxSize, Integer.MAX_VALUE);
        var txnLogBufferedWriter = txnLogBufferedWriterContext.txnLogBufferedWriter;
        // Add some data.
        for (int i = 0; i < writeCount; i++){
            txnLogBufferedWriter.asyncAddData(i, callbackWithCounter.callback, i);
        }

        // Wait for all data write finish.
        Awaitility.await().atMost(2, TimeUnit.SECONDS).until(
                () -> {
                    return callbackWithCounter.failureCounter.get() == writeCount;
                }
        );
        assertEquals(txnLogBufferedWriterContext.mockedManagedLedger.writeCounter.get(), 0);
        // cleanup.
        releaseTxnLogBufferedWriterContext(txnLogBufferedWriterContext);
    }