[pulsar] branch branch-2.10 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 branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/branch-2.10 by this push:
     new 82550b106bd [fix][broker] Fix RawReader out of order (#16390)
82550b106bd is described below

commit 82550b106bd0ad4ba354454a6fbea620b3da0ae9
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);
         }
     }