topicLevelPoliciesEnabled

Don't we need to check if system topic is enabled?

PersistentTopic persistentTopic = new PersistentTopic(topic, ledger, BrokerService.this, isSystemTopic(topic));

final

same comments as above.

define this constant in common module.

same comments as above

If we are using Guava cache, how do we backfill the cache if the entries are evicted out of the cache?

Topic policies service will check the reader is have no more events can read before get topic policies, so when a reader evicted, the topic policies service will create a new reader and cache it.

the two get calls can potentially return different values. so as a good practice, I would recommend writing in the following way.

CompletableFuture<Optional<LookupResult>> future = targetMap.get(bundle);
if (future != null && !future.isDone()) {
    // ...
}

shouldn't we return the future here, since there is already an ongoing lookup, no?

I am actually wondering if it is actually clearer if we create a class SystemTopic extending PersistentTopic. So it will be:

class SystemTopic extends PersistentTopic {
}

PersistentTopic persistentTopic;
if (isSystemTopic(topic)) {
    persistentTopic = new PersistentTopic(topic, ledger, BrokerService.this);
} else {
    persistentTopic = new SystemTopic(topic, ledger, BrokerService.this);
}

Can we move this method to PersistentTopic?

so the logic can be clear:

class PersistentTopic {

    public boolean isBacklogExceed() {
         // check the backlog logic
    }
}

class SystemTopic {
    public boolean isBacklogExceed() {
         return false;
    } 
}

so the method can be simplified as 

public boolean isBacklogExceeded(PersistentTopic topic) {
    return topic.isBacklogExceeded();
}

see my comment above about making a new SystemTopic class and making it extend PersistentTopic.

any reason why do you change this field here?