[fix][broker] Topic policy can not be work well if replay policy message has any exception. #20613

zhanghaou · 2023-06-20T02:37:10Z

Motivation

If listener.onUpdate has any exception, the future can not be complete. We should catch the exception and make the method more safety, otherwise thread would be block.

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

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

Matching PR in forked repository

PR in forked repository:

Technoboy- · 2023-06-21T02:23:46Z

...ker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

-                    }
-                }));
+                    }));
+                } catch (Exception e) {


Throwable is better

mattisonchao · 2023-06-25T02:38:42Z

...ker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

+                    policiesCache.forEach(((topicName, topicPolicies) -> {
+                        if (listeners.get(topicName) != null) {
+                            for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
+                                listener.onUpdate(topicPolicies);


Can we easily catch the onUpdate method's exception with the warn log and then normally complete the future?

IMO, Because it is an implementation problem(bug) in the listener logic, we can't let one bad apple spoils the whole bunch. :)

If we complete directly, it may introduce data inconsistency？ Complete directly or throws exception depends on the importance of this method. From the source pr(#11136) of this code, I think this method is very important.

If the listener throws an exception, it should be a bug. Do you have any ability to auto-fix it by retry?

We need to fix this bug instead of ignoring it.

mattisonchao · 2023-06-25T02:42:35Z

...ker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

+                    }));
+                } catch (Throwable e) {
+                    log.error("[{}] Failed to replay policy message.", reader.getSystemTopic().getTopicName(), e);
+                    future.completeExceptionally(e);


The callback will invoke before cleanCacheAndCloseReader, which may cause some inconsistent status. :)

Considering someone who checks the cache and reader status in the future callback.

Considering completeExceptionally throw exception. (logically)

cleanCacheAndCloseReader had delete from here.

poorbarcode · 2023-06-25T04:05:01Z

...ker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

+                } catch (Throwable e) {
+                    log.error("[{}] Failed to replay policy message.", reader.getSystemTopic().getTopicName(), e);
+                    future.completeExceptionally(e);
+                    cleanCacheAndCloseReader(reader.getSystemTopic().getTopicName().getNamespaceObject(), false);


We should not close the reader here because the caller will close it.

yes, right. We shouldn't close the reader here

cleanCacheAndCloseReader had delete from here.

Technoboy-

We shouldn't close the reader there

zhanghaou · 2023-06-25T06:22:20Z

We shouldn't close the reader there

cleanCacheAndCloseReader had delete from here. And I will make a new pr to fix some other use case.

zhanghaou · 2023-06-25T07:00:52Z

We shouldn't close the reader there

cleanCacheAndCloseReader had delete from here. And I will make a new pr to fix some other use case.

#20640

Technoboy- · 2023-06-26T09:15:59Z

I think we need to process like this :

policiesCache.forEach(((topicName, topicPolicies) -> {
                    if (listeners.get(topicName) != null) {
                        for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
                            try {
                                listener.onUpdate(topicPolicies);
                            } catch (Throwable error) {
                                log.error("[{}] call listener error.", topicName, error);
                            }
                        }
                    }
                }));

zhanghaou · 2023-06-26T10:28:00Z

I think we need to process like this :

policiesCache.forEach(((topicName, topicPolicies) -> {
                    if (listeners.get(topicName) != null) {
                        for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
                            try {
                                listener.onUpdate(topicPolicies);
                            } catch (Throwable error) {
                                log.error("[{}] call listener error.", topicName, error);
                            }
                        }
                    }
                }));

why?

Technoboy- · 2023-06-26T12:04:20Z

I think we need to process like this :

policiesCache.forEach(((topicName, topicPolicies) -> {
                    if (listeners.get(topicName) != null) {
                        for (TopicPolicyListener<TopicPolicies> listener : listeners.get(topicName)) {
                            try {
                                listener.onUpdate(topicPolicies);
                            } catch (Throwable error) {
                                log.error("[{}] call listener error.", topicName, error);
                            }
                        }
                    }
                }));

why?

If the listener has an error, the reader cannot be closed here, because the reader is at the namespace level. Nor can other listeners not receive notifications because of a problem with one topic listener. So, only log output can be printed here

zhanghaou · 2023-06-26T13:30:04Z

@Technoboy- Makes sense. Thanks for your explanation. I also caught exceptions in the notifyListener() method to prevent a listener callback from failing to terminate the remaining listener callbacks. Please take a look again.

codecov-commenter · 2023-06-29T11:00:10Z

Codecov Report

Merging #20613 (2871cc9) into master (2b01f83) will increase coverage by 3.33%.
The diff coverage is 22.64%.

@@             Coverage Diff              @@
##             master   #20613      +/-   ##
============================================
+ Coverage     33.58%   36.92%   +3.33%     
+ Complexity    12127    12118       -9     
============================================
  Files          1613     1690      +77     
  Lines        126241   129192    +2951     
  Branches      13770    14091     +321     
============================================
+ Hits          42396    47698    +5302     
+ Misses        78331    75232    -3099     
- Partials       5514     6262     +748

Flag	Coverage Δ
inttests	`24.14% <19.81%> (-0.05%)`	⬇️
systests	`25.05% <16.98%> (?)`
unittests	`32.03% <56.75%> (-0.01%)`	⬇️

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

Impacted Files	Coverage Δ
...ava/org/apache/pulsar/admin/cli/CmdNamespaces.java	`0.00% <0.00%> (ø)`
...in/java/org/apache/pulsar/admin/cli/CmdTopics.java	`0.00% <0.00%> (ø)`
.../java/org/apache/pulsar/client/cli/CmdProduce.java	`25.90% <9.09%> (-0.76%)`	⬇️
.../service/SystemTopicBasedTopicPoliciesService.java	`57.14% <50.00%> (+0.87%)`	⬆️
...roker/service/persistent/MessageDeduplication.java	`40.17% <50.00%> (ø)`
...he/pulsar/client/impl/MultiTopicsConsumerImpl.java	`32.81% <50.00%> (+0.44%)`	⬆️
.../pulsar/broker/stats/BrokerOperabilityMetrics.java	`100.00% <100.00%> (ø)`
...org/apache/pulsar/broker/stats/DimensionStats.java	`85.45% <100.00%> (ø)`

... and 400 files with indirect coverage changes

…age has any exception. (#20613)

…age has any exception. (#20613) (cherry picked from commit 200fb56)

…age has any exception. (apache#20613) (cherry picked from commit 200fb56) (cherry picked from commit 0a69f66)

…age has any exception. (#20613) (cherry picked from commit 200fb56)

github-actions bot added the doc-not-needed Your PR changes do not impact docs label Jun 20, 2023

zhanghaou changed the title ~~[fix][broker] Topic policy can not be work well if replay policy message has any excetion.~~ [fix][broker] Topic policy can not be work well if replay policy message has any exception. Jun 20, 2023

Technoboy- reviewed Jun 21, 2023

View reviewed changes

Technoboy- added this to the 3.1.0 milestone Jun 21, 2023

Technoboy- assigned zhanghaou Jun 21, 2023

Technoboy- added area/broker release/2.11.2 release/3.0.2 labels Jun 21, 2023

Technoboy- approved these changes Jun 21, 2023

View reviewed changes

mattisonchao reviewed Jun 25, 2023

View reviewed changes

poorbarcode reviewed Jun 25, 2023

View reviewed changes

Technoboy- requested changes Jun 25, 2023

View reviewed changes

zhanghaou mentioned this pull request Jun 25, 2023

[cleanup][broker] remove unneeded use of cleanCacheAndCloseReader #20640

Closed

14 tasks

coderzc approved these changes Jun 27, 2023

View reviewed changes

coderzc requested review from Technoboy-, mattisonchao and poorbarcode June 27, 2023 01:35

coderzc added the release/2.10.5 label Jun 27, 2023

Technoboy- added the ready-to-test label Jun 27, 2023

Technoboy- closed this Jun 27, 2023

Technoboy- reopened this Jun 27, 2023

Technoboy- approved these changes Jun 27, 2023

View reviewed changes

catch exception for replay policy message.

83a8aa3

张浩 added 3 commits June 29, 2023 13:00

use Throwable

c1f849b

remove cleanCacheAndCloseReader.

86a98b2

address comment.

2871cc9

Technoboy- force-pushed the fix-thread-block branch from 6237502 to 2871cc9 Compare June 29, 2023 05:00

Technoboy- closed this Jun 29, 2023

Technoboy- reopened this Jun 29, 2023

Technoboy- merged commit 200fb56 into apache:master Jun 30, 2023
85 of 89 checks passed

Technoboy- pushed a commit that referenced this pull request Jul 3, 2023

[fix][broker] Topic policy can not be work well if replay policy mess…

8f5aa1e

…age has any exception. (#20613)

Technoboy- added cherry-picked/branch-3.0 cherry-picked/branch-2.11 labels Jul 3, 2023

Technoboy- pushed a commit that referenced this pull request Jul 4, 2023

[fix][broker] Topic policy can not be work well if replay policy mess…

6f96d68

…age has any exception. (#20613)

liangyepianzhou pushed a commit that referenced this pull request Jul 7, 2023

[fix][broker] Topic policy can not be work well if replay policy mess…

0a69f66

…age has any exception. (#20613) (cherry picked from commit 200fb56)

liangyepianzhou added the cherry-picked/branch-2.10 label Jul 7, 2023

RobertIndie added release/3.0.1 and removed release/3.0.2 labels Jul 13, 2023

nicoloboschi pushed a commit to datastax/pulsar that referenced this pull request Jul 17, 2023

[fix][broker] Topic policy can not be work well if replay policy mess…

ffe393d

…age has any exception. (apache#20613) (cherry picked from commit 200fb56) (cherry picked from commit 0a69f66)

coderzc added the release/2.9.5 label Aug 22, 2023

coderzc pushed a commit that referenced this pull request Aug 22, 2023

[fix][broker] Topic policy can not be work well if replay policy mess…

b1137e8

…age has any exception. (#20613) (cherry picked from commit 200fb56)

coderzc added the cherry-picked/branch-2.9 Archived: 2.9 is end of life label Aug 22, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[fix][broker] Topic policy can not be work well if replay policy message has any exception. #20613

[fix][broker] Topic policy can not be work well if replay policy message has any exception. #20613

zhanghaou commented Jun 20, 2023

Technoboy- Jun 21, 2023

zhanghaou Jun 21, 2023

mattisonchao Jun 25, 2023

zhanghaou Jun 25, 2023

mattisonchao Jun 25, 2023

zhanghaou Jun 25, 2023

mattisonchao Jun 25, 2023 •

edited

Loading

zhanghaou Jun 25, 2023

poorbarcode Jun 25, 2023

Technoboy- Jun 25, 2023

zhanghaou Jun 25, 2023

Technoboy- left a comment

zhanghaou commented Jun 25, 2023

zhanghaou commented Jun 25, 2023

Technoboy- commented Jun 26, 2023

zhanghaou commented Jun 26, 2023

Technoboy- commented Jun 26, 2023

zhanghaou commented Jun 26, 2023

codecov-commenter commented Jun 29, 2023

[fix][broker] Topic policy can not be work well if replay policy message has any exception. #20613

[fix][broker] Topic policy can not be work well if replay policy message has any exception. #20613

Conversation

zhanghaou commented Jun 20, 2023

Motivation

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mattisonchao Jun 25, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Technoboy- left a comment

Choose a reason for hiding this comment

zhanghaou commented Jun 25, 2023

zhanghaou commented Jun 25, 2023

Technoboy- commented Jun 26, 2023

zhanghaou commented Jun 26, 2023

Technoboy- commented Jun 26, 2023

zhanghaou commented Jun 26, 2023

codecov-commenter commented Jun 29, 2023

Codecov Report

mattisonchao Jun 25, 2023 •

edited

Loading