[improve][txn] PIP-160 admin api coordinatorStats append batch info

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/v3/AdminApiTransactionTest.java

@@ -56,6 +56,7 @@
 import org.apache.pulsar.common.policies.data.TransactionInBufferStats;
 import org.apache.pulsar.common.policies.data.TransactionInPendingAckStats;
 import org.apache.pulsar.common.policies.data.TransactionMetadata;
+import org.apache.pulsar.common.policies.data.TxnBufferedWriterStat;
 import org.apache.pulsar.common.stats.PositionInPendingAckStats;
 import org.apache.pulsar.packages.management.core.MockedPackagesStorageProvider;
 import org.apache.pulsar.transaction.coordinator.impl.MLTransactionLogImpl;
@@ -81,13 +82,25 @@
 @Test(groups = "broker-admin")
 public class AdminApiTransactionTest extends MockedPulsarServiceBaseTest {
 
+    /**
+     * Transaction log batch Configuration.
+     */
+    private boolean transactionBatchEnabled = true;
+    private int transactionMaxCountInBatch = 256;
+    private int transactionBatchMaxSize = 1024 * 1024;
+    private int transactionBatchMaxDelayMillis = 128;
+
     @BeforeMethod
     @Override
     protected void setup() throws Exception {
         conf.setEnablePackagesManagement(true);
         conf.setPackagesManagementStorageProvider(MockedPackagesStorageProvider.class.getName());
         conf.setTransactionCoordinatorEnabled(true);
         conf.setTransactionBufferSnapshotMaxTransactionCount(1);
+        conf.setTransactionLogBatchedWriteEnabled(transactionBatchEnabled);
+        conf.setTransactionLogBatchedWriteMaxRecords(transactionMaxCountInBatch);
+        conf.setTransactionLogBatchedWriteMaxSize(transactionBatchMaxSize);
+        conf.setTransactionLogBatchedWriteMaxDelayInMillis(transactionBatchMaxDelayMillis);
         super.internalSetup();
         admin.clusters().createCluster("test", ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build());
         TenantInfoImpl tenantInfo = new TenantInfoImpl(Sets.newHashSet("role1", "role2"), Sets.newHashSet("test"));
@@ -116,17 +129,20 @@ public void testGetTransactionCoordinatorStats() throws Exception {
         transactionCoordinatorstats = admin.transactions().getCoordinatorStatsByIdAsync(0).get();
         verifyCoordinatorStats(transactionCoordinatorstats.state,
                 transactionCoordinatorstats.leastSigBits, transactionCoordinatorstats.lowWaterMark);
+        verifyTxnBufferedWriterStat(transactionCoordinatorstats.bufferedWriterStat);
         Map<Integer, TransactionCoordinatorStats> stats = admin.transactions().getCoordinatorStatsAsync().get();
 
         assertEquals(stats.size(), 2);
 
         transactionCoordinatorstats = stats.get(0);
         verifyCoordinatorStats(transactionCoordinatorstats.state,
                 transactionCoordinatorstats.leastSigBits, transactionCoordinatorstats.lowWaterMark);
+        verifyTxnBufferedWriterStat(transactionCoordinatorstats.bufferedWriterStat);
 
         transactionCoordinatorstats = stats.get(1);
         verifyCoordinatorStats(transactionCoordinatorstats.state,
                 transactionCoordinatorstats.leastSigBits, transactionCoordinatorstats.lowWaterMark);
+        verifyTxnBufferedWriterStat(transactionCoordinatorstats.bufferedWriterStat);
     }
 
     @Test(timeOut = 20000)
@@ -795,6 +811,14 @@ private static void verifyCoordinatorStats(String state,
         assertEquals(lowWaterMark, 0);
     }
 
+
+    private void verifyTxnBufferedWriterStat(TxnBufferedWriterStat txnBufferedWriterStat) {
+        assertEquals(txnBufferedWriterStat.isBatchEnabled(), transactionBatchEnabled);
+        assertEquals(txnBufferedWriterStat.getBatchedWriteMaxRecords(), transactionMaxCountInBatch);
+        assertEquals(txnBufferedWriterStat.getBatchedWriteMaxSize(), transactionBatchMaxSize);
+        assertEquals(txnBufferedWriterStat.getBatchedWriteMaxDelayInMillis(), transactionBatchMaxDelayMillis);
+    }
+
     private void initTransaction(int coordinatorSize) throws Exception {
         pulsar.getPulsarResources()
                 .getNamespaceResources()

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/TransactionCoordinatorStats.java

@@ -43,4 +43,7 @@ public class TransactionCoordinatorStats {
     public long recoverStartTime;
     //End timestamp of transaction coordinator recovery. 0L means no startup.
     public long recoverEndTime;
+
+    /** The stat of transaction log buffered writer. **/
+    public TxnBufferedWriterStat bufferedWriterStat;
 }

pulsar-client-admin-api/src/main/java/org/apache/pulsar/common/policies/data/TxnBufferedWriterStat.java

@@ -0,0 +1,21 @@
+package org.apache.pulsar.common.policies.data;
+
+import lombok.Data;
+
+@Data
+public class TxnBufferedWriterStat {
+
+    /** Whether to enable the batch feature when writing to Bookie. **/
+    private boolean batchEnabled = false;
+
+    /** If enabled the feature-batch, this attribute means maximum log records count in a batch. **/
+    private int batchedWriteMaxRecords = 512;
+
+    /** If enabled the feature-batch, this attribute means bytes size in a batch. **/
+    private int batchedWriteMaxSize;
+
+    /**
+     * If enabled the feature-batch, this attribute means maximum wait time(in millis) for the first record in a batch.
+     */
+    private int batchedWriteMaxDelayInMillis;
+}

pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionLogImpl.java

@@ -29,6 +29,7 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicLong;
+import lombok.Getter;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedCursor;
@@ -76,6 +77,7 @@ public class MLTransactionLogImpl implements TransactionLog {
 
     private final TopicName topicName;
 
+    @Getter
     private TxnLogBufferedWriter<TransactionMetadataEntry> bufferedWriter;
 
     private final Timer timer;

pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/MLTransactionMetadataStore.java

@@ -453,6 +453,7 @@ public TransactionCoordinatorStats getCoordinatorStats() {
         transactionCoordinatorstats.ongoingTxnSize = txnMetaMap.size();
         transactionCoordinatorstats.recoverStartTime = recoverTime.getRecoverStartTime();
         transactionCoordinatorstats.recoverEndTime = recoverTime.getRecoverEndTime();
+        transactionCoordinatorstats.bufferedWriterStat = transactionLog.getBufferedWriter().getStats();
         return transactionCoordinatorstats;
     }
 

pulsar-transaction/coordinator/src/main/java/org/apache/pulsar/transaction/coordinator/impl/TxnLogBufferedWriter.java

@@ -41,6 +41,7 @@
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
+import org.apache.pulsar.common.policies.data.TxnBufferedWriterStat;
 
 /***
  * See PIP-160: https://github.com/apache/pulsar/issues/15516.
@@ -590,4 +591,13 @@ public void addFailed(ManagedLedgerException exception, Object ctx) {
             }
         }
     }
+
+    public TxnBufferedWriterStat getStats(){
+        TxnBufferedWriterStat txnBufferedWriterStat = new TxnBufferedWriterStat();
+        txnBufferedWriterStat.setBatchEnabled(batchEnabled);
+        txnBufferedWriterStat.setBatchedWriteMaxRecords(batchedWriteMaxRecords);
+        txnBufferedWriterStat.setBatchedWriteMaxSize(batchedWriteMaxSize);
+        txnBufferedWriterStat.setBatchedWriteMaxDelayInMillis(batchedWriteMaxDelayInMillis);
+        return txnBufferedWriterStat;
+    }
 }

site2/docs/admin-api-transactions.md

@@ -417,7 +417,13 @@ The following is an example of the returned values.
   "lowWaterMark" : 0,
   "ongoingTxnSize" : 0,
   "recoverStartTime" : 1657021892377,
-  "recoverEndTime" : 1657021892378
+  "recoverEndTime" : 1657021892378,
+  "bufferedWriterStat":{
+    "batchEnabled": true,
+    "batchedWriteMaxRecords": 512
+    "batchedWriteMaxSize": 4194304,
+    "batchedWriteMaxDelayInMillis": 1
+  }
 }
 
 ```