Make TransactionBuffer's reader use poolmessage to reduce the GC pres…

…sure.
apache · Dec 25, 2024 · 52ffe82 · 52ffe82
1 parent 51e8247
commit 52ffe82
Show file tree

Hide file tree

Showing 5 changed files with 25 additions and 8 deletions.
diff --git a/...ava/org/apache/pulsar/broker/systopic/TransactionBufferSnapshotBaseSystemTopicClient.java b/...ava/org/apache/pulsar/broker/systopic/TransactionBufferSnapshotBaseSystemTopicClient.java
@@ -205,6 +205,7 @@ protected CompletableFuture<Reader<T>> newReaderAsyncInternal() {
                 .subscriptionRolePrefix(SystemTopicNames.SYSTEM_READER_PREFIX)
                 .startMessageId(MessageId.earliest)
                 .readCompacted(true)
+                .poolMessages(true)
                 .createAsync()
                 .thenApply(reader -> {
                     if (log.isDebugEnabled()) {

diff --git a/.../apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java b/.../apache/pulsar/broker/transaction/buffer/impl/SnapshotSegmentAbortedTxnProcessorImpl.java
@@ -768,8 +768,16 @@ private CompletableFuture<Void> clearAllSnapshotSegments() {
                     try {
                         while (wait(reader.hasMoreEventsAsync(), "has more events")) {
                             final var message = wait(reader.readNextAsync(), "read next");
-                            if (topic.getName().equals(message.getValue().getTopicName())) {
-                                snapshotSegmentsWriter.getFuture().get().write(message.getKey(), null);
+                            final String topicName;
+                            final String key;
+                            try {
+                                topicName = message.getValue().getTopicName();
+                                key = message.getKey();
+                            } finally {
+                                message.release();
+                            }
+                            if (topic.getName().equals(topicName)) {
+                                snapshotSegmentsWriter.getFuture().get().write(key, null);
                             }
                         }
                         future.complete(null);

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TableView.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TableView.java
@@ -61,12 +61,16 @@ public T readLatest(String topic) throws Exception {
         final var reader = getReader(topic);
         while (wait(reader.hasMoreEventsAsync(), "has more events")) {
             final var msg = wait(reader.readNextAsync(), "read message");
-            if (msg.getKey() != null) {
-                if (msg.getValue() != null) {
-                    snapshots.put(msg.getKey(), msg.getValue());
-                } else {
-                    snapshots.remove(msg.getKey());
+            try {
+                if (msg.getKey() != null) {
+                    if (msg.getValue() != null) {
+                        snapshots.put(msg.getKey(), msg.getValue());
+                    } else {
+                        snapshots.remove(msg.getKey());
+                    }
                 }
+            } finally {
+                msg.release();
             }
         }
         return snapshots.get(topic);

diff --git a/...er/src/test/java/org/apache/pulsar/broker/transaction/SegmentAbortedTxnProcessorTest.java b/...er/src/test/java/org/apache/pulsar/broker/transaction/SegmentAbortedTxnProcessorTest.java
@@ -348,6 +348,7 @@ private void verifySnapshotSegmentsSize(String topic, int size) throws Exception
                         .createReader(TopicName.get(topic)).get();
         int segmentCount = 0;
         while (reader.hasMoreEvents()) {
+            @Cleanup("release")
             Message<TransactionBufferSnapshotSegment> message = reader.readNextAsync()
                     .get(5, TimeUnit.SECONDS);
             if (topic.equals(message.getValue().getTopicName())) {
@@ -364,6 +365,7 @@ private void verifySnapshotSegmentsIndexSize(String topic, int size) throws Exce
                         .createReader(TopicName.get(topic)).get();
         int indexCount = 0;
         while (reader.hasMoreEvents()) {
+            @Cleanup("release")
             Message<TransactionBufferSnapshotIndexes> message = reader.readNextAsync()
                     .get(5, TimeUnit.SECONDS);
             if (topic.equals(message.getValue().getTopicName())) {

diff --git a/...src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java b/...src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java
@@ -715,7 +715,9 @@ public void testTransactionBufferIndexSystemTopic() throws Exception {
         indexesWriter.write(SNAPSHOT_INDEX, transactionBufferTransactionBufferSnapshotIndexes);
 
         assertTrue(indexesReader.hasMoreEvents());
-        transactionBufferTransactionBufferSnapshotIndexes = indexesReader.readNext().getValue();
+        @Cleanup("release")
+        Message<TransactionBufferSnapshotIndexes> message = indexesReader.readNext();
+        transactionBufferTransactionBufferSnapshotIndexes = message.getValue();
         assertEquals(transactionBufferTransactionBufferSnapshotIndexes.getTopicName(), SNAPSHOT_INDEX);
         assertEquals(transactionBufferTransactionBufferSnapshotIndexes.getIndexList().size(), 5);
         assertNull(transactionBufferTransactionBufferSnapshotIndexes.getSnapshot());