Return-Path: <commits-return-2118-archive-asf-public=cust-asf.ponee.io@pulsar.incubator.apache.org>
X-Original-To: archive-asf-public-internal@cust-asf2.ponee.io
Delivered-To: archive-asf-public-internal@cust-asf2.ponee.io
Received: from cust-asf.ponee.io (cust-asf.ponee.io [163.172.22.183])
	by cust-asf2.ponee.io (Postfix) with ESMTP id 50B12200D5A
	for <archive-asf-public-internal@cust-asf2.ponee.io>; Thu, 14 Dec 2017 15:24:08 +0100 (CET)
Received: by cust-asf.ponee.io (Postfix)
	id 4F18E160C04; Thu, 14 Dec 2017 14:24:08 +0000 (UTC)
Delivered-To: archive-asf-public@cust-asf.ponee.io
Received: from mail.apache.org (hermes.apache.org [140.211.11.3])
	by cust-asf.ponee.io (Postfix) with SMTP id 9589B160C01
	for <archive-asf-public@cust-asf.ponee.io>; Thu, 14 Dec 2017 15:24:07 +0100 (CET)
Received: (qmail 74679 invoked by uid 500); 14 Dec 2017 14:24:06 -0000
Mailing-List: contact commits-help@pulsar.incubator.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:commits-help@pulsar.incubator.apache.org>
List-Unsubscribe: <mailto:commits-unsubscribe@pulsar.incubator.apache.org>
List-Post: <mailto:commits@pulsar.incubator.apache.org>
List-Id: <commits.pulsar.incubator.apache.org>
Reply-To: dev@pulsar.incubator.apache.org
Delivered-To: mailing list commits@pulsar.incubator.apache.org
Received: (qmail 74670 invoked by uid 99); 14 Dec 2017 14:24:06 -0000
Received: from ec2-52-202-80-70.compute-1.amazonaws.com (HELO gitbox.apache.org) (52.202.80.70)
    by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 14 Dec 2017 14:24:06 +0000
From: GitBox <git@apache.org>
To: commits@pulsar.apache.org
Subject: [GitHub] lovelle commented on a change in pull request #963: Partitioned Consumer UnAckedMessageTracker
Message-ID: <151326144625.21220.17818919746417258920.gitbox@gitbox.apache.org>
archived-at: Thu, 14 Dec 2017 14:24:08 -0000

lovelle commented on a change in pull request #963: Partitioned Consumer UnAckedMessageTracker
URL: https://github.com/apache/incubator-pulsar/pull/963#discussion_r156956190
 
 

 ##########
 File path: pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedConsumerImpl.java
 ##########
 @@ -418,23 +449,23 @@ private ConsumerConfiguration getInternalConsumerConfig() {
 
     @Override
     public void redeliverUnacknowledgedMessages() {
-        for (ConsumerImpl c : consumers) {
-            c.redeliverUnacknowledgedMessages();
+        synchronized (this) {
+            incomingMessages.clear();
+            unAckedMessageTracker.clear();
+            for (ConsumerImpl c : consumers) {
+                c.redeliverUnacknowledgedMessages();
+            }
         }
+
     }
 
     @Override
     public void redeliverUnacknowledgedMessages(Set<MessageIdImpl> messageIds) {
+        removeExpiredMessagesFromQueue(messageIds);
+        Map<Integer, Set<MessageIdImpl>> setMap = messageIds.stream()
+                .collect(Collectors.groupingBy(MessageIdImpl::getPartitionIndex, Collectors.toSet()));
         for (ConsumerImpl c : consumers) {
-            Set<MessageIdImpl> consumerMessageIds = new HashSet<>();
-            messageIds.removeIf(messageId -> {
-                if (messageId.getPartitionIndex() == c.getPartitionIndex()) {
-                    consumerMessageIds.add(messageId);
-                    return true;
-                }
-                return false;
-            });
-            c.redeliverUnacknowledgedMessages(consumerMessageIds);
+            c.redeliverUnacknowledgedMessages(setMap.get(c.getPartitionIndex()));
 
 Review comment:
   I believe that these code could lead to a bug.
   
   setMap might not have expired messages for all partitions, therefore `setMap.get(c.getPartitionIndex()))` is going to be `null` generating some possible `NullPointerException` on `redeliverUnacknowledgedMessages` method.
   I imagine two possible solutions:
   1. instead of using `.get` use `.getOrDefault` checking if default was obtained before calling `redeliverUnacknowledgedMessages` method.
   2. Just check if setMap has proper partition index.
   ```java
   int cPartitionIdx = c.getPartitionIndex();
   if (setMap.containsKey(cPartitionIdx)) {
       c.redeliverUnacknowledgedMessages(setMap.get(cPartitionIdx));
   }
   ```
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services