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 testMetricsStatsThatTriggeredByMaxSize() throws Exception {
        SumStrDataSerializer dataSerializer = new SumStrDataSerializer();
        int batchedWriteMaxSize = 16;
        int writeCount = 100;
        int expectedBatchFlushCount = writeCount / (batchedWriteMaxSize / dataSerializer.getSizePerData());
        int expectedTotalBytesSize = expectedBatchFlushCount * batchedWriteMaxSize;
        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, "");
        }
        // Wait for all write finish.
        Awaitility.await().atMost(1, TimeUnit.SECONDS).until(
                () -> callbackWithCounter.finishCounter.get() + callbackWithCounter.failureCounter.get() == writeCount
        );
        int actualBatchFlushCount = txnLogBufferedWriterContext.mockedManagedLedger.writeCounter.get();
        assertEquals(callbackWithCounter.failureCounter.get(), 0);
        assertEquals(expectedBatchFlushCount, actualBatchFlushCount);
        verifyTheCounterMetrics(0, expectedBatchFlushCount,0,0);
        verifyTheHistogramMetrics(expectedBatchFlushCount, writeCount, expectedTotalBytesSize);
        // cleanup.
        releaseTxnLogBufferedWriterContext(txnLogBufferedWriterContext);
        // after close, verify the metrics change to 0.
        verifyTheCounterMetrics(0,0,0,0);
        verifyTheHistogramMetrics(0,0,0);
    }