WHAT

We enforce partition limits through the CreateTopicPolicy and Authorizer plugin, but we should still allow partitions for internal topics to be created even if the limit has been reached. We already discount internal partitions when counting the number of existing partitions, but new internal partition requests could be blocked if the limit has been reached.

WHY

Even if a cluster has reached the advertised/enforced limit for customer partitions, we may want to increase the number of internal partitions (either as new internal topics, or by growing existing internal topics) to support the running of the service. If we instead subject our internal partitions to the enforcement, then for some clusters our ability to effectively manage the cluster may be reduced, and we'd have clusters with accidental differences in configuration.

HOW

Suggestion:

• Update ManagedKafkaCreateTopicPolicy#validateNumPartitions to return early successfully if the topic being created has an internal prefix.
• The CustomAclAuthorizer#authorizeAction method should perform the super user check before the partition limit checking, and should exclude internal partitions from the partition limit checking (if possible).

NOTES

1. Once this change has been rolled out, we may need to be a little more careful when increasing the number of internal partitions. It would be good if components that increase the number of internal partitions would first ensure that their actions would not cause the total partition count to be above the absolute cluster limit, so that we don't cause availability issues for customers. I don't know how possible or enforceable this would be though.