[BUG] Catch topic policy not hit exception in handleSubscribe by hangc0276 · Pull Request #10341 · apache/pulsar

hangc0276 · 2021-04-23T09:30:51Z

Motivation

When running KOP with topic policy, it get the following exception.

17:09:46.777 [mock-pulsar-bk:org.apache.pulsar.broker.service.ServerCnx@1010] WARN  org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:55980][persistent://pulsar/test/localhost:15000/__change_events-partition-0][multiTopicsReader-c3d8054591] Failed to create consumer: consumerId=0, Topic policies cache have not init.
java.util.concurrent.CompletionException: org.apache.pulsar.broker.service.BrokerServiceException$TopicPoliciesCacheNotInitException: Topic policies cache have not init.
        at java.util.concurrent.CompletableFuture.encodeRelay(CompletableFuture.java:326) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture.completeRelay(CompletableFuture.java:338) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture.uniRelay(CompletableFuture.java:911) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:953) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:632) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_172]
        at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_172]
        at org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.lambda$addEntry$32(BookkeeperSchemaStorage.java:566) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT]
        at org.apache.bookkeeper.client.PulsarMockLedgerHandle.lambda$asyncAddEntry$5(PulsarMockLedgerHandle.java:194) ~[testmocks-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT]
        at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) [?:1.8.0_172]
        at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) [?:1.8.0_172]
        at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442) [?:1.8.0_172]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_172]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_172]
        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_172]
Caused by: org.apache.pulsar.broker.service.BrokerServiceException$TopicPoliciesCacheNotInitException: Topic policies cache have not init.
        at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.getTopicPolicies(SystemTopicBasedTopicPoliciesService.java:148) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT]
        at org.apache.pulsar.broker.service.persistent.PersistentTopic.checkSubscriptionTypesEnable(PersistentTopic.java:2872) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT]
        at org.apache.pulsar.broker.service.persistent.PersistentTopic.subscribe(PersistentTopic.java:648) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT]
        at org.apache.pulsar.broker.service.ServerCnx.lambda$null$12(ServerCnx.java:966) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT]
        at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952) ~[?:1.8.0_172]
        ... 15 more
17:09:46.801 [pulsar-client-io-34-1:org.apache.pulsar.client.impl.ClientCnx@682] WARN  org.apache.pulsar.client.impl.ClientCnx - [id: 0xb9d000f5, L:/127.0.0.1:55980 - R:localhost/127.0.0.1:15002] Received error from server: Topic policies cache have not init.
17:09:46.804 [pulsar-client-io-34-1:org.apache.pulsar.client.impl.ConsumerImpl@770] WARN  org.apache.pulsar.client.impl.ConsumerImpl - [persistent://pulsar/test/localhost:15000/__change_events-partition-0][multiTopicsReader-c3d8054591] Failed to subscribe to topic on localhost/127.0.0.1:15002
17:09:46.804 [pulsar-client-io-34-1:org.apache.pulsar.client.impl.ConnectionHandler@102] WARN  org.apache.pulsar.client.impl.ConnectionHandler - [persistent://pulsar/test/localhost:15000/__change_events-partition-0] [multiTopicsReader-c3d8054591] Could not get connection to broker: Topic policies cache have not init. -- Will try again in 0.1 s

The reason is that it doesn't catch exception in getTopicPolicies, which will lead to subscribe failed.

public boolean checkSubscriptionTypesEnable(SubType subType) throws Exception {
        TopicName topicName = TopicName.get(topic);
        if (brokerService.pulsar().getConfiguration().isTopicLevelPoliciesEnabled()) {
            TopicPolicies topicPolicies =
                    brokerService.pulsar().getTopicPoliciesService().getTopicPolicies(TopicName.get(topic));
            if (topicPolicies == null) {
                return checkNsAndBrokerSubscriptionTypesEnable(topicName, subType);
            } else {
                if (topicPolicies.getSubscriptionTypesEnabled().isEmpty()) {
                    return checkNsAndBrokerSubscriptionTypesEnable(topicName, subType);
                }
                return topicPolicies.getSubscriptionTypesEnabled().contains(subType);
            }
        } else {
            return checkNsAndBrokerSubscriptionTypesEnable(topicName, subType);
        }
    }

Changes

catch the exception in checkSubscriptionTypesEnable
expose exception stack in servercnx#handleSubscribe

eolivelli · 2021-04-23T10:06:59Z

Is It possibile to add a test case?

315157973

LGTM

codelipenghui

The change looks good to me, could you please help add a test to avoid regression

merlimat · 2021-04-24T00:44:38Z

                                        log.warn("[{}][{}][{}] Failed to create consumer: consumerId={}, {}",
                                                 remoteAddress, topicName, subscriptionName,
-                                                 consumerId,  exception.getCause().getMessage());
+                                                 consumerId,  exception.getCause().getMessage(), exception);


This was done on purpose to avoid stack traces for "known" or expected exceptions.

sijie · 2021-05-04T02:51:45Z

@hangc0276 Can you address Matteo's comment?

hangc0276 · 2021-05-10T12:20:52Z

@hangc0276 Can you address Matteo's comment?

@sijie Ok, i will address it.

### Modification 1. upgrade pulsar dependency to include the following bug fix - apache/pulsar#10706 : fix load __consumer_offset topic timeout on bundle onLoad stage - apache/pulsar#10341 : fix create reader failed on bundle onLoad stage

…#10341) ### Motivation When running KOP with topic policy, it get the following exception. ``` 17:09:46.777 [mock-pulsar-bk:org.apache.pulsar.broker.service.ServerCnx@1010] WARN org.apache.pulsar.broker.service.ServerCnx - [/127.0.0.1:55980][persistent://pulsar/test/localhost:15000/__change_events-partition-0][multiTopicsReader-c3d8054591] Failed to create consumer: consumerId=0, Topic policies cache have not init. java.util.concurrent.CompletionException: org.apache.pulsar.broker.service.BrokerServiceException$TopicPoliciesCacheNotInitException: Topic policies cache have not init. at java.util.concurrent.CompletableFuture.encodeRelay(CompletableFuture.java:326) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture.completeRelay(CompletableFuture.java:338) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture.uniRelay(CompletableFuture.java:911) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:953) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture$UniCompose.tryFire(CompletableFuture.java:926) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture.uniAccept(CompletableFuture.java:656) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture$UniAccept.tryFire(CompletableFuture.java:632) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:474) ~[?:1.8.0_172] at java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:1962) ~[?:1.8.0_172] at org.apache.pulsar.broker.service.schema.BookkeeperSchemaStorage.lambda$addEntry$32(BookkeeperSchemaStorage.java:566) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT] at org.apache.bookkeeper.client.PulsarMockLedgerHandle.lambda$asyncAddEntry$5(PulsarMockLedgerHandle.java:194) ~[testmocks-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT] at java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:760) [?:1.8.0_172] at java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:736) [?:1.8.0_172] at java.util.concurrent.CompletableFuture$Completion.run(CompletableFuture.java:442) [?:1.8.0_172] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_172] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_172] at java.lang.Thread.run(Thread.java:748) [?:1.8.0_172] Caused by: org.apache.pulsar.broker.service.BrokerServiceException$TopicPoliciesCacheNotInitException: Topic policies cache have not init. at org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService.getTopicPolicies(SystemTopicBasedTopicPoliciesService.java:148) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT] at org.apache.pulsar.broker.service.persistent.PersistentTopic.checkSubscriptionTypesEnable(PersistentTopic.java:2872) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT] at org.apache.pulsar.broker.service.persistent.PersistentTopic.subscribe(PersistentTopic.java:648) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT] at org.apache.pulsar.broker.service.ServerCnx.lambda$null$12(ServerCnx.java:966) ~[pulsar-broker-2.8.0-SNAPSHOT.jar:2.8.0-SNAPSHOT] at java.util.concurrent.CompletableFuture.uniCompose(CompletableFuture.java:952) ~[?:1.8.0_172] ... 15 more 17:09:46.801 [pulsar-client-io-34-1:org.apache.pulsar.client.impl.ClientCnx@682] WARN org.apache.pulsar.client.impl.ClientCnx - [id: 0xb9d000f5, L:/127.0.0.1:55980 - R:localhost/127.0.0.1:15002] Received error from server: Topic policies cache have not init. 17:09:46.804 [pulsar-client-io-34-1:org.apache.pulsar.client.impl.ConsumerImpl@770] WARN org.apache.pulsar.client.impl.ConsumerImpl - [persistent://pulsar/test/localhost:15000/__change_events-partition-0][multiTopicsReader-c3d8054591] Failed to subscribe to topic on localhost/127.0.0.1:15002 17:09:46.804 [pulsar-client-io-34-1:org.apache.pulsar.client.impl.ConnectionHandler@102] WARN org.apache.pulsar.client.impl.ConnectionHandler - [persistent://pulsar/test/localhost:15000/__change_events-partition-0] [multiTopicsReader-c3d8054591] Could not get connection to broker: Topic policies cache have not init. -- Will try again in 0.1 s ``` The reason is that it doesn't catch exception in getTopicPolicies, which will lead to subscribe failed. ```Java public boolean checkSubscriptionTypesEnable(SubType subType) throws Exception { TopicName topicName = TopicName.get(topic); if (brokerService.pulsar().getConfiguration().isTopicLevelPoliciesEnabled()) { TopicPolicies topicPolicies = brokerService.pulsar().getTopicPoliciesService().getTopicPolicies(TopicName.get(topic)); if (topicPolicies == null) { return checkNsAndBrokerSubscriptionTypesEnable(topicName, subType); } else { if (topicPolicies.getSubscriptionTypesEnabled().isEmpty()) { return checkNsAndBrokerSubscriptionTypesEnable(topicName, subType); } return topicPolicies.getSubscriptionTypesEnabled().contains(subType); } } else { return checkNsAndBrokerSubscriptionTypesEnable(topicName, subType); } } ``` ### Changes 1. catch the exception in `checkSubscriptionTypesEnable` 2. expose exception stack in `servercnx#handleSubscribe`

catch topic policy not hit exception in handleSubscribe

868c218

hangc0276 requested review from 315157973, codelipenghui, eolivelli and sijie April 23, 2021 09:31

hangc0276 self-assigned this Apr 23, 2021

315157973 approved these changes Apr 23, 2021

View reviewed changes

codelipenghui reviewed Apr 23, 2021

View reviewed changes

merlimat reviewed Apr 24, 2021

View reviewed changes

BewareMyPower mentioned this pull request Apr 26, 2021

Add inner topic protection streamnative/kop#451

Merged

sijie changed the title ~~【BUG】Catch topic policy not hit exception in handleSubscribe~~ [BUG] Catch topic policy not hit exception in handleSubscribe May 4, 2021

hangc0276 added release/blocker Indicate the PR or issue that should block the release until it gets resolved type/bug The PR fixed a bug or issue reported a bug labels May 27, 2021

hangc0276 added this to the 2.8.0 milestone May 27, 2021

add test

79e89d9

hangc0276 requested review from codelipenghui and merlimat May 27, 2021 05:09

codelipenghui approved these changes May 27, 2021

View reviewed changes

codelipenghui merged commit 608e36e into apache:master May 27, 2021

hangc0276 mentioned this pull request May 30, 2021

upgrade pulsar dependency to 2.8.0-rc-202105291235 streamnative/kop#535

Merged

Navigation Menu

Repository navigation

[BUG] Catch topic policy not hit exception in handleSubscribe - #10341

[BUG] Catch topic policy not hit exception in handleSubscribe#10341
codelipenghui merged 2 commits into
apache:masterfrom
hangc0276:chenhang/fix_topic_policy_not_init_exception

hangc0276 commented Apr 23, 2021

Uh oh!

eolivelli commented Apr 23, 2021 •

edited

Loading

Uh oh!

315157973 left a comment

Uh oh!

codelipenghui left a comment

Uh oh!

merlimat Apr 24, 2021

Uh oh!

sijie commented May 4, 2021

Uh oh!

hangc0276 commented May 10, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

Conversation

hangc0276 commented Apr 23, 2021

Motivation

Changes

Uh oh!

eolivelli commented Apr 23, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

315157973 left a comment

Choose a reason for hiding this comment

Uh oh!

codelipenghui left a comment

Choose a reason for hiding this comment

Uh oh!

merlimat Apr 24, 2021

Choose a reason for hiding this comment

Uh oh!

sijie commented May 4, 2021

Uh oh!

hangc0276 commented May 10, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

eolivelli commented Apr 23, 2021 •

edited

Loading