[pulsar] branch master updated: [fix][broker] Fix RawReader out of order (#16390)

[mm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 76646cfd949 [fix][broker] Fix RawReader out of order (#16390)
76646cfd949 is described below

commit 76646cfd9496820a6a6dbee225067045235838c0
Author: Hang Chen <ch...@apache.org>
AuthorDate: Thu Jul 7 01:54:22 2022 +0800

    [fix][broker] Fix RawReader out of order (#16390)
    
    * fix RawReader out ouf order
    
    * address comments
    
    * tune code
---
 .../src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java   | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
index e738e5916b4..92cbbe4a399 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
@@ -211,9 +211,10 @@ public class RawReaderImpl implements RawReader {
                 log.debug("[{}][{}] Received raw message: {}/{}/{}", topic, subscription,
                         messageId.getEntryId(), messageId.getLedgerId(), messageId.getPartition());
             }
+
             incomingRawMessages.add(
-                    new RawMessageAndCnx(new RawMessageImpl(messageId, headersAndPayload), cnx));
-            tryCompletePending();
+                new RawMessageAndCnx(new RawMessageImpl(messageId, headersAndPayload), cnx));
+            internalPinnedExecutor.execute(this::tryCompletePending);
         }
     }