[fix][broker] Fix consume stuck of shared streaming dispatcher #18315

coderzc · 2022-11-03T07:41:28Z

Motivation

In #18289, we set sendInProgress to false only after the last entry in a batch of messages has been delivered, but this will make consume stuck, because StreamingEntryReader.pendingReads may is not empty when without available entries from the ledger and lead to ctx.isLast() always be false.

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java

Lines 172 to 177 in 82237d3

    
           // All request has been completed, mark returned entry as last. 
        
           if (issuedReads.isEmpty() && pendingReads.isEmpty()) { 
        
               firstPendingReadEntryRequest.isLast = true; 
        
               STATE_UPDATER.set(this, State.Completed); 
        
           } 
        
           dispatcher.readEntryComplete(readEntry, firstPendingReadEntryRequest);

Modifications

Abstract acquireSendInProgress()/releaseSendInProgress()/isSendInProgress()
In subclass PersistentStreamingDispatcherMultipleConsumers, use sendingTaskCounter instead of sendInProgress to avoid sendInProgress to be set to true and false multiple times.
Clean up useless code

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: coderzc#26

coderzc · 2022-11-04T06:45:12Z

@poorbarcode PTAL

poorbarcode

#18314 is designed to solve the problem of multiple set sendInProgress true caused by normal type reads.

Instead isLast, we should use another identifier to indicate that this entry is the last entry to be read：

E.g 1: the last entry in queue issuedReads
E.g 2: the previous entry of pendingReads.peek

Both of these implementations should be somewhat difficult because the two queues are constantly changing

poorbarcode · 2022-11-04T09:03:10Z

.../apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java

            dispatchMessagesThread.execute(safeRun(() -> {
-                if (sendMessagesToConsumers(readType, Lists.newArrayList(entry), ctx.isLast())) {
+                if (sendMessagesToConsumers(readType, Lists.newArrayList(entry),
+                        readType == ReadType.Normal || ctx.isLast())) {


Only normal-type reads will run here, so readType == ReadType.Normal will always be true.

You are right, I try to separate the sendInProgress and sendInProgressReplay, make sendInProgress of normal type read don't affect sendInProgressReplay of replay type read. Please look at it again.

codecov-commenter · 2022-11-07T03:17:01Z

Codecov Report

Merging #18315 (5045881) into master (b31c5a6) will decrease coverage by 0.82%.
The diff coverage is 40.44%.

@@             Coverage Diff              @@
##             master   #18315      +/-   ##
============================================
- Coverage     46.98%   46.16%   -0.83%     
+ Complexity    10343     9936     -407     
============================================
  Files           692      667      -25     
  Lines         67766    65699    -2067     
  Branches       7259     7029     -230     
============================================
- Hits          31842    30332    -1510     
+ Misses        32344    31946     -398     
+ Partials       3580     3421     -159

Flag	Coverage Δ
unittests	`46.16% <40.44%> (-0.83%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
.../org/apache/bookkeeper/mledger/impl/EntryImpl.java	`69.62% <0.00%> (-11.27%)`	⬇️
...lsar/broker/service/RedeliveryTrackerDisabled.java	`50.00% <ø> (ø)`
...va/org/apache/pulsar/broker/service/ServerCnx.java	`48.91% <ø> (+0.22%)`	⬆️
...ersistentStreamingDispatcherMultipleConsumers.java	`0.00% <0.00%> (ø)`
.../java/org/apache/pulsar/client/impl/ClientCnx.java	`30.16% <ø> (ø)`
...a/org/apache/pulsar/client/impl/TableViewImpl.java	`0.00% <0.00%> (ø)`
...ar/client/impl/conf/ProducerConfigurationData.java	`84.70% <ø> (-0.18%)`	⬇️
...va/org/apache/pulsar/client/impl/ConsumerImpl.java	`15.07% <12.50%> (+0.03%)`	⬆️
...keeper/mledger/impl/cache/RangeEntryCacheImpl.java	`45.79% <22.44%> (-6.31%)`	⬇️
...eeper/mledger/impl/cache/InflightReadsLimiter.java	`30.35% <30.35%> (ø)`
... and 93 more

poorbarcode · 2022-11-07T03:33:32Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

    protected boolean shouldRewindBeforeReadingOrReplaying = false;
    protected final String name;
    protected boolean sendInProgress;
+    protected boolean sendInProgressReplay;


Maybe we can combine the two variables into a counter? Just like this:

AtomicInteger sendingTaskCounter; sendingTaskCounter.increment(); try { trySendMessagesToConsumer(); } finally { int runningTaskCount = sendingTaskCounter.decrementAndGet(); if (runningTaskCount == 0){ readMoreEntries(); } }

I think this makes the logic easier to unstand.

Good idea, but seems to need to remove sendInProgress=false in the PersistentStickyKeyDispatcherMultipleConsumers, I use delay execute readMoreEntries to replace it.

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

poorbarcode · 2022-11-09T08:39:42Z

.../apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

            // stuckConsumers for avoid stopping dispatch.
-            sendInProgress = false;
-            topic.getBrokerService().executor().execute(safeRun(this::readMoreEntries));
+            topic.getBrokerService().executor().schedule(safeRun(this::readMoreEntries), 100, TimeUnit.MILLISECONDS);


Why the 100ms delay?

because sendingTaskCounter will be decrement after the execution trySendMessagesToConsumers, I think it can replace sendInProgress = false in the line-325

readMoreEntries is triggered in the method sendMessagesToConsumers， can we remove these two lines?

topic.getBrokerService().executor().schedule(safeRun(this::readMoreEntries), 100, TimeUnit.MILLISECONDS);

It seems that it doesn't need to be delayed execution, the trySendMessagesToConsumers is also within the synchronized block. I want to can be directly removed sendInProgress = false.

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

poorbarcode · 2022-11-09T09:51:12Z

@codelipenghui @congbobo184 @Technoboy- @mattisonchao Could you take a look?

poorbarcode · 2022-11-09T09:58:59Z

.../apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

            // stuckConsumers for avoid stopping dispatch.
-            sendInProgress = false;
-            topic.getBrokerService().executor().execute(safeRun(this::readMoreEntries));
+            topic.getBrokerService().executor().schedule(safeRun(this::readMoreEntries), 100, TimeUnit.MILLISECONDS);


readMoreEntries is triggered in the method sendMessagesToConsumers， can we remove these two lines?

topic.getBrokerService().executor().schedule(safeRun(this::readMoreEntries), 100, TimeUnit.MILLISECONDS);

…reamingDispatcherMultipleConsumers

coderzc · 2022-11-11T03:15:31Z

@poorbarcode @codelipenghui PTAL

codelipenghui · 2022-11-11T10:22:31Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+    }
+
+    protected final synchronized void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
+        java.util.function.Consumer<Boolean> consumer = asyncRead -> {


Is it possible to avoid creating a new java.util.function.Consumer for each consumer's message dispatching?

codelipenghui · 2022-11-11T10:23:13Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+       sendMessagesToConsumers(readType, entries);
+    }
+
+    protected synchronized void sendInProgressAcquire() {


nit:

Suggested change

protected synchronized void sendInProgressAcquire() {

protected synchronized void acquireSendInProgress() {

codelipenghui · 2022-11-11T10:23:30Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+        sendInProgress = true;
+    }
+
+    protected synchronized void sendInProgressRelease() {


nit:

Suggested change

protected synchronized void sendInProgressRelease() {

protected synchronized void releaseSendInProgress() {

codelipenghui · 2022-11-11T10:23:48Z

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

+        sendInProgress = false;
+    }
+
+    protected synchronized boolean sendInProgress() {


nit:

Suggested change

protected synchronized boolean sendInProgress() {

protected synchronized boolean isSendInProgress() {

# Conflicts: # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java # pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java

poorbarcode · 2022-11-14T01:32:56Z

.../apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java

            // readMoreEntries should run regardless whether or not stuck is caused by
            // stuckConsumers for avoid stopping dispatch.
-            sendInProgress = false;
            topic.getBrokerService().executor().execute(safeRun(this::readMoreEntries));


Should we remove this tow line(line-325, line-327) and make this method return true at these two logic branch? Then readMoreEntries will be triggered by super.readEntriesComplete.

You are right.

poorbarcode

LGTM

coderzc · 2022-11-14T08:46:01Z

/pulsarbot run-failure-checks

coderzc · 2022-11-15T02:55:59Z

@codelipenghui PTAL

fix consume stuck of shared streaming dispatcher

3346851

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Nov 3, 2022

codelipenghui assigned coderzc Nov 3, 2022

codelipenghui added this to the 2.12.0 milestone Nov 3, 2022

codelipenghui added the type/flaky-tests label Nov 3, 2022

codelipenghui approved these changes Nov 3, 2022

View reviewed changes

poorbarcode approved these changes Nov 4, 2022

View reviewed changes

poorbarcode requested changes Nov 4, 2022

View reviewed changes

separate sendInProgress and sendInProgressReplay

02163c5

poorbarcode reviewed Nov 7, 2022

View reviewed changes

Use sendingTaskCounter instead of sendInProgress

d0ce918

poorbarcode reviewed Nov 9, 2022

View reviewed changes

poorbarcode mentioned this pull request Nov 9, 2022

[fix][test] fix flaky test PersistentTopicStreamingDispatcherE2ETest.testMessageRedelivery #18378

Closed

2 tasks

poorbarcode approved these changes Nov 9, 2022

View reviewed changes

.../java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java Outdated Show resolved Hide resolved

add synchronized for haveSendingTask

6cb3908

poorbarcode approved these changes Nov 9, 2022

View reviewed changes

improve code

2a05e08

coderzc force-pushed the fix_stream_stuck branch from 7c39d19 to 2a05e08 Compare November 9, 2022 10:32

aloyszhang approved these changes Nov 10, 2022

View reviewed changes

Use sendingTaskCounter instead of sendInProgress only in PersistentSt…

50a0ae1

…reamingDispatcherMultipleConsumers

codelipenghui reviewed Nov 11, 2022

View reviewed changes

coderzc added 3 commits November 11, 2022 19:28

Address comment

1530839

rebase master

b57decc

poorbarcode reviewed Nov 14, 2022

View reviewed changes

remove useless code

5045881

poorbarcode approved these changes Nov 14, 2022

View reviewed changes

codelipenghui merged commit 55863de into apache:master Nov 18, 2022

Technoboy- pushed a commit that referenced this pull request May 24, 2023

[fix][broker] fix consume stuck of shared streaming dispatcher (#18315)

fec0fca

Technoboy- added area/broker cherry-picked/branch-2.11 release/2.11.2 labels May 24, 2023

Technoboy- changed the title ~~[fix][broker] fix consume stuck of shared streaming dispatcher~~ [fix][broker] Fix consume stuck of shared streaming dispatcher Jul 18, 2023

lhotari mentioned this pull request Aug 26, 2024

[improve][broker] Reschedule reads with increasing backoff when no messages are dispatched #23226

Merged

4 tasks

	// All request has been completed, mark returned entry as last.
	if (issuedReads.isEmpty() && pendingReads.isEmpty()) {
	firstPendingReadEntryRequest.isLast = true;
	STATE_UPDATER.set(this, State.Completed);
	}
	dispatcher.readEntryComplete(readEntry, firstPendingReadEntryRequest);

	protected synchronized void sendInProgressAcquire() {
	protected synchronized void acquireSendInProgress() {

	protected synchronized void sendInProgressRelease() {
	protected synchronized void releaseSendInProgress() {

	protected synchronized boolean sendInProgress() {
	protected synchronized boolean isSendInProgress() {

[fix][broker] Fix consume stuck of shared streaming dispatcher #18315

[fix][broker] Fix consume stuck of shared streaming dispatcher #18315

Uh oh!

Conversation

coderzc commented Nov 3, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Documentation

Matching PR in forked repository

Uh oh!

coderzc commented Nov 4, 2022

Uh oh!

poorbarcode left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

coderzc Nov 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codecov-commenter commented Nov 7, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

coderzc Nov 9, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

poorbarcode commented Nov 9, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

coderzc commented Nov 11, 2022

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poorbarcode Nov 14, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poorbarcode left a comment

Choose a reason for hiding this comment

Uh oh!

coderzc commented Nov 14, 2022

Uh oh!

coderzc commented Nov 15, 2022

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

coderzc commented Nov 3, 2022 •

edited

Loading

coderzc Nov 7, 2022 •

edited

Loading

codecov-commenter commented Nov 7, 2022 •

edited

Loading

coderzc Nov 9, 2022 •

edited

Loading

poorbarcode Nov 14, 2022 •

edited

Loading