[incubator-pulsar] branch master updated: Fix race in MultiTopicsConsumerImpl#redeliverUnacknowledgedMessages (#1952)

[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/incubator-pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new dbfe9b7  Fix race in MultiTopicsConsumerImpl#redeliverUnacknowledgedMessages (#1952)
dbfe9b7 is described below

commit dbfe9b7fc352f3691a5b35ece74727a696fc6d8a
Author: Ivan Kelly <iv...@apache.org>
AuthorDate: Wed Jun 13 19:28:23 2018 +0200

    Fix race in MultiTopicsConsumerImpl#redeliverUnacknowledgedMessages (#1952)
    
    This method calls #redeliverUnacknowledgedMessages on all underlying
    consumers and clears the incomingMessages queue. There was a race that
    the redelivery could happen before the incomingMessages was cleared,
    and therefore redelivered messages could be lost. Previously this
    block was protected by synchronized(this) block, but this didn't
    protect against any other blocks.
    
    This patch changes the synchronization to use the writeLock of the
    same lock which is acquired before messages are pushed onto the queue.
    
    This race was the source of flake in
    ResendRequestTest.testSharedSingleAckedPartitionedTopic
    
    Issue: #1160
---
 .../org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java     | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index e7d3b26..4be8f58 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -502,12 +502,15 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
 
     @Override
     public void redeliverUnacknowledgedMessages() {
-        synchronized (this) {
+        lock.writeLock().lock();
+        try {
             consumers.values().stream().forEach(consumer -> consumer.redeliverUnacknowledgedMessages());
             incomingMessages.clear();
             unAckedMessageTracker.clear();
-            resumeReceivingFromPausedConsumersIfNeeded();
+        } finally {
+            lock.writeLock().unlock();
         }
+        resumeReceivingFromPausedConsumersIfNeeded();
     }
 
     @Override

-- 
To stop receiving notification emails like this one, please contact
mmerli@apache.org.