Repository: apache/kafka
File: core/src/main/java/kafka/server/QuotaFactory.java (Java)

Code before:
import org.apache.kafka.server.quota.ControllerMutationQuotaManager;
import org.apache.kafka.server.quota.QuotaType;

import java.util.Optional;


public class QuotaFactory {

    public static final ReplicaQuota UNBOUNDED_QUOTA = new ReplicaQuota() {
        @Override
        public boolean isThrottled(TopicPartition topicPartition) {
            return false;
        }

        @Override
        public boolean isQuotaExceeded() {
            return false;
        }

        @Override
        public void record(long value) {
            // No-op
        }
    };

    public static class QuotaManagers {
        private final ClientQuotaManager fetch;
        private final ClientQuotaManager produce;
        private final ClientRequestQuotaManager request;
        private final ControllerMutationQuotaManager controllerMutation;
        private final ReplicationQuotaManager leader;
        private final ReplicationQuotaManager follower;
        private final ReplicationQuotaManager alterLogDirs;
        private final Optional<Plugin<ClientQuotaCallback>> clientQuotaCallbackPlugin;

        public QuotaManagers(ClientQuotaManager fetch, ClientQuotaManager produce, ClientRequestQuotaManager request,
                             ControllerMutationQuotaManager controllerMutation, ReplicationQuotaManager leader,
                             ReplicationQuotaManager follower, ReplicationQuotaManager alterLogDirs,
                             Optional<Plugin<ClientQuotaCallback>> clientQuotaCallbackPlugin) {
            this.fetch = fetch;
            this.produce = produce;
            this.request = request;
            this.controllerMutation = controllerMutation;
            this.leader = leader;
            this.follower = follower;
            this.alterLogDirs = alterLogDirs;
            this.clientQuotaCallbackPlugin = clientQuotaCallbackPlugin;
        }

        public ClientQuotaManager fetch() {
            return fetch;

Review comment (bug):
I prefer not to use `record`, as mutable fields can lead to undefined behavior in `equals` and `hashCode`

Code after:
import org.apache.kafka.server.quota.ControllerMutationQuotaManager;
import org.apache.kafka.server.quota.QuotaType;

import java.util.Optional;


public class QuotaFactory {

    public static final ReplicaQuota UNBOUNDED_QUOTA = new ReplicaQuota() {
        @Override
        public boolean isThrottled(TopicPartition topicPartition) {
            return false;
        }

        @Override
        public boolean isQuotaExceeded() {
            return false;
        }

        @Override
        public void record(long value) {
            // No-op
        }
    };

    public record QuotaManagers(ClientQuotaManager fetch,
                                ClientQuotaManager produce,
                                ClientRequestQuotaManager request,
                                ControllerMutationQuotaManager controllerMutation,
                                ReplicationQuotaManager leader,
                                ReplicationQuotaManager follower,
                                ReplicationQuotaManager alterLogDirs,
                                Optional<Plugin<ClientQuotaCallback>> clientQuotaCallbackPlugin) {

        public void shutdown() {
            fetch.shutdown();
            produce.shutdown();
            request.shutdown();
            controllerMutation.shutdown();
            clientQuotaCallbackPlugin.ifPresent(plugin -> Utils.closeQuietly(plugin, "client quota callback plugin"));
        }
    }

    public static QuotaManagers instantiate(
        KafkaConfig cfg,
        Metrics metrics,
        Time time,
        String threadNamePrefix,
        String role
    ) {
        Optional<Plugin<ClientQuotaCallback>> clientQuotaCallbackPlugin = createClientQuotaCallback(cfg, metrics, role);