[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #15107: [fix] [client-consumer] Fix message discard problem when pending receives all done.

[GitBox <gi...@apache.org>]

michaeljmarshall commented on code in PR #15107:
URL: https://github.com/apache/pulsar/pull/15107#discussion_r847777124


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1155,16 +1156,24 @@ protected <V> MessageImpl<V> newMessage(final MessageIdImpl messageId,
     }
 
     private void executeNotifyCallback(final MessageImpl<T> message) {
+        internalPinnedExecutor.execute(() -> {
+            executeNotifyCallback0(message);
+        });
+    }
+
+    @VisibleForTesting
+    void executeNotifyCallback0(final MessageImpl<T> message) {

Review Comment:
   Instead of exposing this internal method for a test, are you able to create a funcational test that verifies this edge case?



##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java:
##########
@@ -1419,40 +1428,38 @@ private ByteBuf processMessageChunk(ByteBuf compressedPayload, MessageMetadata m
      * Notify waiting asyncReceive request with the received message.
      *
      * @param message
+     * @return the message is enqueue incomingMessages.
      */
-    void notifyPendingReceivedCallback(final Message<T> message, Exception exception) {
-        if (pendingReceives.isEmpty()) {

Review Comment:
   Can you clarify which race you're referencing? If you're referring to the implementation of `hasNextPendingReceive()`, `pendingReceives` is only supposed to be updated on the `internalPinnedExecutor`, so the check is redundant, but is not necessarily a race. (We can still remove the check, but I just want to see if there is another race you noticed.)



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org