You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/11/16 05:01:19 UTC

[GitHub] [pulsar] poorbarcode opened a new pull request, #18491: [do not merge] [fix] [client] Multi topic consumer receive null even if has backlog

poorbarcode opened a new pull request, #18491:
URL: https://github.com/apache/pulsar/pull/18491

   Fixes #18480
   
   - #18480
   
   ### Motivation
   
   TODO add description
   
   ### Modifications
   
   TODO add description and improve code
   
   
   ### Documentation
   
   <!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
   
   - [ ] `doc` <!-- Your PR contains doc changes. Please attach the local preview screenshots (run `sh start.sh` at `pulsar/site2/website`) to your PR description, or else your PR might not get merged. -->
   - [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
   - [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
   - [ ] `doc-complete` <!-- Docs have been already added -->
   
   ### Matching PR in forked repository
   
   PR in forked repository: 
   - https://github.com/poorbarcode/pulsar/pull/40
   


-- 
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

[GitHub] [pulsar] BewareMyPower merged pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
BewareMyPower merged PR #18491:
URL: https://github.com/apache/pulsar/pull/18491


-- 
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

[GitHub] [pulsar] poorbarcode commented on pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
poorbarcode commented on PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#issuecomment-1316964954

   Hi @BewareMyPower 
   > And in my local env, `resumeReceivingFromPausedConsumersIfNeeded` and `clearIncomingMessages` were all executed in the same thread:
   
   if `resumeReceivingFromPausedConsumersIfNeeded` and `clearIncomingMessages` runs in the same thread, the cmd `resumeReceivingFromPausedConsumersIfNeeded` will always executed before `clearIncomingMessages`, this leads to a higher probability of problems.


-- 
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

[GitHub] [pulsar] BewareMyPower commented on pull request #18491: [fix] [client] Multi topic consumer receive null even if has backlog

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#issuecomment-1316675708

   Could you also explain why the flaky test is caused by this reason? I added some logs:
   
   ```diff
   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/im
   pl/MultiTopicsConsumerImpl.java
   index 1853be55f9..809a590451 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
   @@ -705,6 +705,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
        @Override
        public void redeliverUnacknowledgedMessages() {
            internalPinnedExecutor.execute(() -> {
   +            log.info("XYZ rededeliverUnacked");
                CONSUMER_EPOCH.incrementAndGet(this);
                consumers.values().stream().forEach(consumer -> {
                    consumer.redeliverUnacknowledgedMessages();
   @@ -713,6 +714,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
                clearIncomingMessages();
                unAckedMessageTracker.clear();
            });
   +        log.info("XYZ resumeReceiving");
            resumeReceivingFromPausedConsumersIfNeeded();
        }
   ```
   
   And in my local env, `resumeReceivingFromPausedConsumersIfNeeded` and `clearIncomingMessages` were all executed in the same thread:
   
   ```
   2022-11-16T17:24:38,714 - INFO  - [pulsar-client-internal-39-1:MultiTopicsConsumerImpl@717] - XYZ resumeReceiving
   2022-11-16T17:24:38,718 - INFO  - [pulsar-client-internal-39-1:MultiTopicsConsumerImpl@708] - XYZ rededeliverUnacked
   ```


-- 
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

[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1024863159


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java:
##########
@@ -439,4 +443,76 @@ public void run() {
         Assert.assertEquals(count, 9);
         Assert.assertEquals(0, datas.size());
     }
+
+    /**
+     * see https://github.com/apache/pulsar/pull/18491
+     */

Review Comment:
   ```suggestion
   ```
   
   Remove the comments



-- 
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

[GitHub] [pulsar] poorbarcode commented on pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
poorbarcode commented on PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#issuecomment-1317283136

   Hi @BewareMyPower 
   
   > > this leads to a higher probability of problems.
   
   > Could you explain why? 
   
   The reason in Menu-Motivation
   
   > Or is there a way to make testFailoverConsumerBatchCumulateAck more likely to fail in local env? This test is hard to fail in local env but easy to fail in CI.
   
   I guess that in CI the incoming queue can be filled quickly enough before the receive has not started.  We can't make it runs faster locally, but we can make receive execution slower, you can change the test like this which makes the test fail in the local env easily:
   
   <img width="917" alt="截屏2022-11-17 00 18 32" src="https://user-images.githubusercontent.com/25195800/202235145-e6a4bf17-e792-4cf1-a81a-ac31d024be32.png">
   
   
   
   


-- 
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

[GitHub] [pulsar] BewareMyPower commented on pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#issuecomment-1318224636

   Got it. Thanks. The motivation part is a little confusing. The root cause is that the consumer won't be removed from `pausedConsumers` if the incoming messages are greater than half of the receiver queue size. Therefore, `resumeReceivingFromPausedConsumersIfNeeded` must be called after `clearIncomingMessages`.
   
   BTW, it's not a race condition in `redeliverUnacknowledgedMessages` for previous code:
   
   ```java
       public void redeliverUnacknowledgedMessages() {
           internalPinnedExecutor.execute(() -> {
               /* ... */
               clearIncomingMessages();
               unAckedMessageTracker.clear();
           });
           resumeReceivingFromPausedConsumersIfNeeded();
       }
   ```
   
   `internalPinnedExecutor` is a single thread executor, when `redeliverUnacknowledgedMessages` is called in `internalPinnedExecutor`, `resumeReceivingFromPausedConsumersIfNeeded` will always be called before `clearIncomingMessages()`.
   


-- 
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

[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1025989421


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   @congbobo184 Is there any more concern? The flaky test has blocked many PRs.



-- 
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

[GitHub] [pulsar] congbobo184 commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
congbobo184 commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1025400068


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   https://github.com/apache/pulsar/blob/dd8bd05ad98fa302d5dc85642ef994c20890e171/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java#L691-L692
   the core issue is: change the lock to `internalPinnedExecutor` then `CONSUMER_EPOCH.incrementAndGet(this)`, the epoch can't work, right?
   
   I'm not sure about the PR `https://github.com/apache/pulsar/pull/18475`, why do we also need to revert? Is the truth the deadlock?
   the fix in `https://github.com/apache/pulsar/pull/18454` i am not sure this fix isn't right.



-- 
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

[GitHub] [pulsar] poorbarcode commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
poorbarcode commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1025418354


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java:
##########
@@ -439,4 +443,76 @@ public void run() {
         Assert.assertEquals(count, 9);
         Assert.assertEquals(0, datas.size());
     }
+
+    /**
+     * see https://github.com/apache/pulsar/pull/18491
+     */

Review Comment:
   Already remove



-- 
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

[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1024855178


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java:
##########
@@ -439,4 +443,76 @@ public void run() {
         Assert.assertEquals(count, 9);
         Assert.assertEquals(0, datas.size());
     }
+
+    /**
+     * see https://github.com/apache/pulsar/pull/18491
+     */
+    @Test
+    public void testMultiTopicConsumerConcurrentRedeliverAndReceive() throws Exception {
+        final String topic = BrokerTestUtil.newUniqueName("my-topic");
+        admin.topics().createPartitionedTopic(topic, 2);
+
+        final int receiverQueueSize = 10;
+
+        @Cleanup
+        MultiTopicsConsumerImpl<Integer> consumer =
+                (MultiTopicsConsumerImpl<Integer>) pulsarClient.newConsumer(Schema.INT32)
+                .topic(topic)
+                .subscriptionName("sub")
+                .receiverQueueSize(receiverQueueSize)
+                .subscribe();
+
+        @Cleanup
+        Producer<Integer> producer = pulsarClient.newProducer(Schema.INT32)
+                .topic(topic)
+                .enableBatching(false)
+                .create();
+
+        for (int i = 0; i < receiverQueueSize; i++){
+            producer.send(i);
+        }
+
+        Awaitility.await().until(() -> consumer.incomingMessages.size() == receiverQueueSize);
+
+        // For testing the race condition of issue #18491
+        // We need to inject a delay for the pinned internal thread
+        injectDelayToInternalThread(consumer, 1000L);
+        consumer.redeliverUnacknowledgedMessages();
+        // Make sure the message redelivery is completed. The incoming queue will be cleaned up during the redelivery.
+        waitForAllTasksForInternalThread(consumer);
+
+        Set<Integer> receivedMsgs = new HashSet<>();
+        for (;;){
+            Message<Integer> msg = consumer.receive(2, TimeUnit.SECONDS);
+            if (msg == null){
+                break;
+            }
+            receivedMsgs.add(msg.getValue());
+        }
+        Assert.assertEquals(receivedMsgs.size(), 10);
+    }
+
+    private void injectDelayToInternalThread(MultiTopicsConsumerImpl<?> consumer, long delayInMillis){
+        ExecutorService internalPinnedExecutor =
+                        WhiteboxImpl.getInternalState(consumer, "internalPinnedExecutor");
+        internalPinnedExecutor.execute(() -> {
+            try {
+                Thread.sleep(delayInMillis);
+            } catch (InterruptedException ignore) {
+            }
+        });
+    }
+
+    /**
+     * If the task after "redeliver" finish, means task-redeliver finish.
+     */
+    private void waitForAllTasksForInternalThread(MultiTopicsConsumerImpl consumer){

Review Comment:
   ```suggestion
       private void waitForAllTasksForInternalThread(MultiTopicsConsumerImpl<?> consumer) {
   ```



-- 
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

[GitHub] [pulsar] codelipenghui commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1023949118


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java:
##########
@@ -439,4 +443,61 @@ public void run() {
         Assert.assertEquals(count, 9);
         Assert.assertEquals(0, datas.size());
     }
+
+    /**
+     * see https://github.com/apache/pulsar/pull/18491
+     */
+    @Test
+    public void testMultiTopicConsumerConcurrentRedeliverAndReceive() throws Exception {

Review Comment:
   Ah, maybe it's better to add a sleep before `receive()` messages from the consumer.



-- 
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

[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18491: [fix] [client] Multi topic consumer receive null even if has backlog

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1023721634


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   In `resumeReceivingFromPausedConsumersIfNeeded`:
   
   ```java
                   internalPinnedExecutor.execute(() -> {
                       receiveMessageFromConsumer(consumer, true);
                   });
   ```
   
   Could you avoid calling `internalPinnedExecutor.execute` when `resumeReceivingFromPausedConsumersIfNeeded` is already called in `internalPinnedExecutor`?



-- 
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

[GitHub] [pulsar] poorbarcode commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
poorbarcode commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1024228893


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   I think it's possible, but not in this PR. A separate PR needs to be set up to optimize task scheduling, and I am happy to do it.



-- 
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

[GitHub] [pulsar] codecov-commenter commented on pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#issuecomment-1319559893

   # [Codecov](https://codecov.io/gh/apache/pulsar/pull/18491?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
   > Merging [#18491](https://codecov.io/gh/apache/pulsar/pull/18491?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (6c391d9) into [master](https://codecov.io/gh/apache/pulsar/commit/79750231051db849350e3d35dd3706320466acac?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7975023) will **decrease** coverage by `8.62%`.
   > The diff coverage is `57.57%`.
   
   [![Impacted file tree graph](https://codecov.io/gh/apache/pulsar/pull/18491/graphs/tree.svg?width=650&height=150&src=pr&token=acYqCpsK9J&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)](https://codecov.io/gh/apache/pulsar/pull/18491?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
   
   ```diff
   @@             Coverage Diff              @@
   ##             master   #18491      +/-   ##
   ============================================
   - Coverage     45.62%   36.99%   -8.63%     
   + Complexity    10075     1965    -8110     
   ============================================
     Files           697      208     -489     
     Lines         68024    14426   -53598     
     Branches       7293     1582    -5711     
   ============================================
   - Hits          31033     5337   -25696     
   + Misses        33413     8508   -24905     
   + Partials       3578      581    -2997     
   ```
   
   | Flag | Coverage Δ | |
   |---|---|---|
   | unittests | `36.99% <57.57%> (-8.63%)` | :arrow_down: |
   
   Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
   
   | [Impacted Files](https://codecov.io/gh/apache/pulsar/pull/18491?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
   |---|---|---|
   | [...he/pulsar/client/impl/MultiTopicsConsumerImpl.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL011bHRpVG9waWNzQ29uc3VtZXJJbXBsLmphdmE=) | `22.26% <0.00%> (ø)` | |
   | [...nt/impl/PulsarClientImplementationBindingImpl.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1B1bHNhckNsaWVudEltcGxlbWVudGF0aW9uQmluZGluZ0ltcGwuamF2YQ==) | `72.41% <ø> (-0.47%)` | :arrow_down: |
   | [...ar/client/impl/conf/ConsumerConfigurationData.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL2NvbmYvQ29uc3VtZXJDb25maWd1cmF0aW9uRGF0YS5qYXZh) | `92.55% <ø> (+2.12%)` | :arrow_up: |
   | [...he/pulsar/client/impl/TypedMessageBuilderImpl.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWNsaWVudC9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2NsaWVudC9pbXBsL1R5cGVkTWVzc2FnZUJ1aWxkZXJJbXBsLmphdmE=) | `30.76% <59.37%> (+3.56%)` | :arrow_up: |
   | [...pache/bookkeeper/mledger/LedgerOffloaderStats.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9MZWRnZXJPZmZsb2FkZXJTdGF0cy5qYXZh) | | |
   | [...che/bookkeeper/mledger/offload/OffloaderUtils.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bWFuYWdlZC1sZWRnZXIvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2Jvb2trZWVwZXIvbWxlZGdlci9vZmZsb2FkL09mZmxvYWRlclV0aWxzLmphdmE=) | | |
   | [.../service/SystemTopicBasedTopicPoliciesService.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1N5c3RlbVRvcGljQmFzZWRUb3BpY1BvbGljaWVzU2VydmljZS5qYXZh) | | |
   | [...ava/org/apache/pulsar/broker/tools/BrokerTool.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci90b29scy9Ccm9rZXJUb29sLmphdmE=) | | |
   | [...lance/extensions/channel/ServiceUnitStateData.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9sb2FkYmFsYW5jZS9leHRlbnNpb25zL2NoYW5uZWwvU2VydmljZVVuaXRTdGF0ZURhdGEuamF2YQ==) | | |
   | [...ulsar/broker/service/PulsarChannelInitializer.java](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL1B1bHNhckNoYW5uZWxJbml0aWFsaXplci5qYXZh) | | |
   | ... and [487 more](https://codecov.io/gh/apache/pulsar/pull/18491/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
   


-- 
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

[GitHub] [pulsar] congbobo184 commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
congbobo184 commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1025400068


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   https://github.com/apache/pulsar/blob/dd8bd05ad98fa302d5dc85642ef994c20890e171/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java#L691-L692
   the core issue is: change the lock to `internalPinnedExecutor` then `CONSUMER_EPOCH.incrementAndGet(this)`, the epoch can't work, right?
   
   I'm not sure about the PR https://github.com/apache/pulsar/pull/18475, why do we also need to revert? Is the truth the deadlock?
   the fix in https://github.com/apache/pulsar/pull/18454 i am not sure this fix isn't right.
   @codelipenghui @poorbarcode @BewareMyPower @Technoboy- @mattisonchao 



-- 
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

[GitHub] [pulsar] poorbarcode commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
poorbarcode commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1024337980


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java:
##########
@@ -439,4 +443,61 @@ public void run() {
         Assert.assertEquals(count, 9);
         Assert.assertEquals(0, datas.size());
     }
+
+    /**
+     * see https://github.com/apache/pulsar/pull/18491
+     */
+    @Test
+    public void testMultiTopicConsumerConcurrentRedeliverAndReceive() throws Exception {

Review Comment:
   Already add a sleep before receiving messages.
   
   I have tried to increase the probability of test failure, but can't get to 100%. Because that:
   - method `receiveMessageFromConsumer` will call `consumerImpl.batchReceiveAsync` and `consumerImpl.receiveAsync` .
   - When `consumerImpl.batchReceiveAsync` and `consumerImpl.receiveAsync` is finished, it will trigger another `receiveMessageFromConsumer`.
   - `consumerImpl.batchReceiveAsync` and `consumerImpl.receiveAsync` calls each other
     - if receive messages is empty call `consumerImpl.receiveAsync`
     - else call `consumerImpl.batchReceiveAsync`
   - `consumerImpl.batchReceiveAsync` and `consumerImpl.receiveAsync` executed asynchronously.



-- 
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

[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1025995690


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   The `CONSUMER_EPOCH` issue is related to https://github.com/apache/pulsar/pull/16160. I think we can fix it in another PR. So I dismissed the review for now.



-- 
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

[GitHub] [pulsar] poorbarcode commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
poorbarcode commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1024231244


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   For multi-topic consumer, there is a lot of optimization to be done



-- 
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

[GitHub] [pulsar] poorbarcode commented on pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
poorbarcode commented on PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#issuecomment-1318872660

   Hi @BewareMyPower 
   
   > We can avoid sleeping by calling redeliverUnacknowledgedMessages in the internal pinned executor. 
   
   Yes, you are right. This is definitely a good idea, and I tried it in the previous [commit](https://github.com/apache/pulsar/pull/18491/commits/399d76ad4ca3f767411b4707a39584c669f15537), but it didn't make more present of failure than `sleep`, so I ended up using sleep instead.
   
   <img width="1288" alt="截屏2022-11-18 00 08 17" src="https://user-images.githubusercontent.com/25195800/202497743-89c076ac-8114-4196-bb0a-a82847bcab33.png">
   
   Here's why：https://github.com/apache/pulsar/pull/18491#discussion_r1024337980
   


-- 
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

[GitHub] [pulsar] codelipenghui commented on a diff in pull request #18491: [fix] [client] Multi topic consumer receive null even if has backlog

Posted by GitBox <gi...@apache.org>.
codelipenghui commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1023655858


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java:
##########
@@ -439,4 +443,61 @@ public void run() {
         Assert.assertEquals(count, 9);
         Assert.assertEquals(0, datas.size());
     }
+
+    /**
+     * see https://github.com/apache/pulsar/pull/18491
+     */
+    @Test
+    public void testMultiTopicConsumerConcurrentRedeliverAndReceive() throws Exception {

Review Comment:
   I have tried the test, but it seems unable to cover the change or reproduce the issue without this fix.
   
   After changing the test like this
   
   ```diff
   diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
   index 56f2d3f29ec..c476cb516aa 100644
   --- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
   +++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java
   @@ -474,9 +474,12 @@ public class NegativeAcksTest extends ProducerConsumerBase {
    
            Awaitility.await().until(() -> consumer.incomingMessages.size() == receiverQueueSize);
    
   +        // For testing the race condition of issue #18491
   +        // We need to inject a delay for the pinned internal thread
   +        injectDelayToInternalThread(consumer, 1000L);
            consumer.redeliverUnacknowledgedMessages();
   -        waitMultiTopicConsumerRedeliverFinish(consumer);
   -
   +        // Make sure the message redelivery is completed. The incoming queue will be cleaned up during the redelivery.
   +        waitForAllTasksForInternalThread(consumer);
            Set<Integer> receivedMsgs = new HashSet<>();
            for (;;){
                Message<Integer> msg = consumer.receive(2, TimeUnit.SECONDS);
   @@ -491,7 +494,18 @@ public class NegativeAcksTest extends ProducerConsumerBase {
        /**
         * If the task after "redeliver" finish, means task-redeliver finish.
         */
   -    private void waitMultiTopicConsumerRedeliverFinish(MultiTopicsConsumerImpl consumer){
   +    private void injectDelayToInternalThread(MultiTopicsConsumerImpl<?> consumer, long delayInMillis){
   +        ExecutorService internalPinnedExecutor =
   +                WhiteboxImpl.getInternalState(consumer, "internalPinnedExecutor");
   +        internalPinnedExecutor.execute(() -> {
   +            try {
   +                Thread.sleep(delayInMillis);
   +            } catch (InterruptedException ignore) {
   +            }
   +        });
   +    }
   +
   +    private void waitForAllTasksForInternalThread(MultiTopicsConsumerImpl<?> consumer) {
            ExecutorService internalPinnedExecutor =
                    WhiteboxImpl.getInternalState(consumer, "internalPinnedExecutor");
            CompletableFuture<Void> taskAfterRedeliver = new CompletableFuture<>();
   ```
   
   The issue can be reproduced without the fix and passed with the fix.



-- 
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

[GitHub] [pulsar] BewareMyPower commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1025454809


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   The fix #18454 is right. But the test added in #18454 accidentally triggerred the bug.



-- 
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

[GitHub] [pulsar] congbobo184 commented on a diff in pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
congbobo184 commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1026005486


##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java:
##########
@@ -712,8 +712,8 @@ public void redeliverUnacknowledgedMessages() {
             });
             clearIncomingMessages();
             unAckedMessageTracker.clear();
+            resumeReceivingFromPausedConsumersIfNeeded();

Review Comment:
   approved 



-- 
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

[GitHub] [pulsar] poorbarcode commented on a diff in pull request #18491: [fix] [client] Multi topic consumer receive null even if has backlog

Posted by GitBox <gi...@apache.org>.
poorbarcode commented on code in PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#discussion_r1023712454


##########
pulsar-broker/src/test/java/org/apache/pulsar/client/impl/NegativeAcksTest.java:
##########
@@ -439,4 +443,61 @@ public void run() {
         Assert.assertEquals(count, 9);
         Assert.assertEquals(0, datas.size());
     }
+
+    /**
+     * see https://github.com/apache/pulsar/pull/18491
+     */
+    @Test
+    public void testMultiTopicConsumerConcurrentRedeliverAndReceive() throws Exception {

Review Comment:
   Yes, this is not a 100% hit test problem, I tried to fix it
   
   <img width="1094" alt="截屏2022-11-16 16 55 42" src="https://user-images.githubusercontent.com/25195800/202136994-1ddca39c-6fbe-4e61-88b5-74c0132f7e08.png">
   



-- 
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

[GitHub] [pulsar] BewareMyPower commented on pull request #18491: [fix][client] Fix multi-topic consumer stuck after redeliver messages

Posted by GitBox <gi...@apache.org>.
BewareMyPower commented on PR #18491:
URL: https://github.com/apache/pulsar/pull/18491#issuecomment-1317201177

   > this leads to a higher probability of problems.
   
   Could you explain why? Or is there a way to make `testFailoverConsumerBatchCumulateAck` more likely to fail in local env? This test is hard to fail in local env but easy to fail in CI.


-- 
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