Move  into future stage to catch the exception
@@ -380,10 +380,13 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
}
private void readMorePolicies(SystemTopicClient.Reader<PulsarEvent> reader) {
- reader.readNextAsync().whenComplete((msg, ex) -> {
- if (ex == null) {
+ reader.readNextAsync()
+ .thenAccept(msg -> {
refreshTopicPoliciesCache(msg);
notifyListener(msg);
+ })
+ .whenComplete((__, ex) -> {
+ if (ex == null) {
readMorePolicies(reader);
} else {
Throwable cause = FutureUtil.unwrapCompletionException(ex);
