[improve][txn][PIP-196]Segmented transaction buffer snapshot configuration

conf/broker.conf

@@ -1497,12 +1497,22 @@ transactionCoordinatorEnabled=false
 transactionMetadataStoreProviderClassName=org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStoreProvider
 
 # Transaction buffer takes a snapshot after the number of transaction operations reaches this value.
+# If transaction buffer enables snapshot segment, transaction buffer updates snapshot metadata
+# after the number of transaction operations reaches this value.
 transactionBufferSnapshotMaxTransactionCount=1000
 
-# Transaction buffer take snapshot interval time
+# Transaction buffer take snapshot interval time.
+# If transaction buffer enables snapshot segment, this is transaction updating snapshot metadata internal time.
 # Unit : millisecond
 transactionBufferSnapshotMinTimeInMillis=5000
 
+# Whether to enable segmented transaction buffer snapshot which is able to handle a large number of aborted transactions.
+transactionBufferSegmentedSnapshotEnabled = false
+
+# Transaction buffer will stored the transaction ID of the aborted transaction, and take snapshot for it.
+# This configuration is configured to determine the size of the snapshot segment. defualt:256kb.
+transactionBufferSnapshotSegmentSize=262144
+
 # The max concurrent requests for transaction buffer client, default is 1000
 transactionBufferClientMaxConcurrentRequests=1000
 

conf/standalone.conf

@@ -1108,12 +1108,22 @@ transactionCoordinatorEnabled=false
 transactionMetadataStoreProviderClassName=org.apache.pulsar.transaction.coordinator.impl.MLTransactionMetadataStoreProvider
 
 # Transaction buffer takes a snapshot after the number of transaction operations reaches this value.
+# If transaction buffer enables snapshot segment, transaction buffer updates snapshot metadata
+# after the number of transaction operations reaches this value.
 transactionBufferSnapshotMaxTransactionCount=1000
 
-# Transaction buffer take snapshot interval time
+# Transaction buffer take snapshot interval time.
+# IF transaction buffer enables snapshot segment, this is transaction updating snapshot metadata internal time.
 # Unit : millisecond
 transactionBufferSnapshotMinTimeInMillis=5000
 
+# Whether to enable segmented transaction buffer snapshot which is able to handle a large number of aborted transactions.
+transactionBufferSegmentedSnapshotEnabled = false
+
+# Transaction buffer will stored the transaction ID of the aborted transaction, and take snapshot for it.
+# This configuration is configured to determine the size of the snapshot segment. defualt:256kb.
+transactionBuffeSnapshotSegmentSize=262144
+
 # The transaction buffer client's operation timeout in milliseconds.
 transactionBufferClientOperationTimeoutInMills=3000
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -2731,15 +2731,34 @@ public class ServiceConfiguration implements PulsarConfiguration {
     @FieldContext(
             category = CATEGORY_TRANSACTION,
             doc = "Transaction buffer take snapshot transaction count"
+                    + "If transaction buffer enables snapshot segment, transaction buffer updates snapshot metadata"
+                    + "after the number of transaction operations reaches this value."
     )
     private int transactionBufferSnapshotMaxTransactionCount = 1000;
 
     @FieldContext(
             category = CATEGORY_TRANSACTION,
             doc = "Transaction buffer take snapshot min interval time"
+                    + "If transaction buffer enables snapshot segment, "
+                    + "this is transaction updating snapshot metadata internal time."
     )
     private int transactionBufferSnapshotMinTimeInMillis = 5000;
 
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Transaction buffer will stored the transaction ID of the aborted transaction, and take snapshot "
+                    + "for it."
+                    + "This configuration is configured to determine the size of the snapshot segment. defualt:256kb."
+    )
+    private int transactionBufferSnapshotSegmentSize = 262144;
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Whether to enable segmented transaction buffer snapshot "
+                    + "which is able to handle a large number of aborted transactions."
+    )
+    private boolean transactionBufferSegmentedSnapshotEnabled = false;
+
     @FieldContext(
             category = CATEGORY_TRANSACTION,
             doc = "The max concurrent requests for transaction buffer client."

pulsar-broker/src/test/java/org/apache/pulsar/common/naming/ServiceConfigurationTest.java

@@ -322,4 +322,25 @@ public void testTransactionBatchConfigurations() throws Exception{
             assertEquals(configuration.getTransactionPendingAckBatchedWriteMaxDelayInMillis(), 20);
         }
     }
+
+    @Test
+    public void testTransactionMultipleSnapshot() throws Exception {
+        ServiceConfiguration configuration = null;
+        // broker.conf.
+        try (FileInputStream inputStream = new FileInputStream("../conf/broker.conf")) {
+            configuration = PulsarConfigurationLoader.create(inputStream, ServiceConfiguration.class);
+            assertEquals(configuration.getTransactionBufferSnapshotSegmentSize(), 262144);
+            assertFalse(configuration.isTransactionBufferSegmentedSnapshotEnabled());
+        }
+        // string input stream.
+        StringBuilder stringBuilder = new StringBuilder();
+        stringBuilder.append("transactionBufferSnapshotSegmentSize=262144").append(System.lineSeparator());
+        stringBuilder.append("transactionBufferSegmentedSnapshotEnabled = false").append(System.lineSeparator());
+        try(ByteArrayInputStream inputStream =
+                    new ByteArrayInputStream(stringBuilder.toString().getBytes(StandardCharsets.UTF_8))){
+            configuration = PulsarConfigurationLoader.create(inputStream, ServiceConfiguration.class);
+            assertEquals(configuration.getTransactionBufferSnapshotSegmentSize(), 262144);
+            assertFalse(configuration.isTransactionBufferSegmentedSnapshotEnabled());
+        }
+    }
 }