Fixed refcounting when processing entries from sql (apache#2316)

willome · Aug 7, 2018 · 6b7a76f · 6b7a76f
1 parent 9213e58
commit 6b7a76f
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 12 deletions.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageParser.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageParser.java
@@ -95,15 +95,10 @@ public static void parseMessage(TopicName topicName, long ledgerId, long entryId
             if (numMessages == 1 && !msgMetadata.hasNumMessagesInBatch()) {
                 final MessageImpl<?> message = new MessageImpl<>(msgId, msgMetadata, uncompressedPayload, null, null);
                 processor.process(msgId, message, uncompressedPayload);
-
-                uncompressedPayload.release();
-
             } else {
                 // handle batch message enqueuing; uncompressed payload has all messages in batch
                 receiveIndividualMessagesFromBatch(msgMetadata, uncompressedPayload, messageId, null, -1, processor);
-                uncompressedPayload.release();
             }
-
         } finally {
             if (uncompressedPayload != null) {
                 uncompressedPayload.release();

diff --git a/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java b/pulsar-sql/presto-pulsar/src/main/java/org/apache/pulsar/sql/presto/PulsarRecordCursor.java
@@ -52,7 +52,6 @@
 import java.util.Map;
 import java.util.Queue;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Consumer;
 
 import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
 import static com.facebook.presto.spi.type.BigintType.BIGINT;
@@ -202,10 +201,9 @@ public boolean advanceNextPosition() {
                 throw new RuntimeException(e);
             }
 
-            newEntries.forEach(new Consumer<Entry>() {
-                @Override
-                public void accept(Entry entry) {
-                    completedBytes += entry.getData().length;
+            newEntries.forEach(entry -> {
+                try {
+                    completedBytes += entry.getDataBuffer().readableBytes();
                     // filter entries that is not part of my split
                     if (((PositionImpl) entry.getPosition()).compareTo(pulsarSplit.getEndPosition()) < 0) {
                         try {
@@ -217,9 +215,9 @@ public void accept(Entry entry) {
                             log.error(e, "Failed to parse message from pulsar topic %s", topicName.toString());
                             throw new RuntimeException(e);
                         }
-                    } else {
-                        entry.release();
                     }
+                } finally {
+                    entry.release();
                 }
             });
         }