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 testMetricsStatsThatTriggeredByLargeSingleData() throws Exception {
        // Use TwoLenSumDataSerializer for: write a little data once, then write a large data once.
        int bytesSizePerRecordWhichInBatch = 4;
        int batchedWriteMaxSize = 1024;
        TwoLenSumDataSerializer dataSerializer =
                new TwoLenSumDataSerializer(bytesSizePerRecordWhichInBatch, batchedWriteMaxSize);
        int writeCount = 100;
        // Both "directly write BK because the data too large" and "flush batch data because next data too large" will
        // write BK, so expectWriteBKCount equals writeCount.
        int expectWriteBKCount = writeCount;
        int singleLargeDataRequestCount = expectWriteBKCount / 2;
        int expectedBatchFlushTriggeredByLargeData = singleLargeDataRequestCount;
        int expectedTotalBytesSize = expectedBatchFlushTriggeredByLargeData * bytesSizePerRecordWhichInBatch;
        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(1, callbackWithCounter.callback, i);
        }
        // Wait for all data write finish.
        Awaitility.await().atMost(2, TimeUnit.SECONDS).until(
                () -> callbackWithCounter.finishCounter.get() + callbackWithCounter.failureCounter.get() == writeCount
        );
        assertEquals(callbackWithCounter.failureCounter.get(), 0);
        assertEquals(txnLogBufferedWriterContext.mockedManagedLedger.writeCounter.get(), expectWriteBKCount);
        verifyTheCounterMetrics(0,0,0,expectedBatchFlushTriggeredByLargeData);
        verifyTheHistogramMetrics(expectedBatchFlushTriggeredByLargeData,
                writeCount - singleLargeDataRequestCount, expectedTotalBytesSize);
        // cleanup.
        releaseTxnLogBufferedWriterContext(txnLogBufferedWriterContext);
        // after close, verify the metrics change to 0.
        verifyTheCounterMetrics(0,0,0,0);
        verifyTheHistogramMetrics(0,0,0);
    }