@Override
    public void configure(Map<String, ?> rawConfigs) {

        Map<String, Object> configs = new HashMap<>(rawConfigs);

        StaticAutoBalancerConfig staticAutoBalancerConfig = new StaticAutoBalancerConfig(configs, false);
        Properties producerProps = AutoBalancerMetricsReporterConfig.parseProducerConfigs(configs);

        // Add BootstrapServers if not set
        if (!producerProps.containsKey(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG)) {
            String bootstrapServers = getBootstrapServers(configs, staticAutoBalancerConfig.getString(StaticAutoBalancerConfig.AUTO_BALANCER_CLIENT_LISTENER_NAME_CONFIG));
            producerProps.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
            LOGGER.info("Using default value of {} for {}", bootstrapServers,
                    AutoBalancerMetricsReporterConfig.config(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
        }

        AutoBalancerMetricsReporterConfig reporterConfig = new AutoBalancerMetricsReporterConfig(configs, false);

        setIfAbsent(producerProps,
                ProducerConfig.CLIENT_ID_CONFIG,
                QuotaConfigs.INTERNAL_CLIENT_ID_PREFIX + reporterConfig.getString(AutoBalancerMetricsReporterConfig.AUTO_BALANCER_METRICS_REPORTER_PRODUCER_CLIENT_ID));
        setIfAbsent(producerProps, ProducerConfig.LINGER_MS_CONFIG,
                reporterConfig.getLong(AutoBalancerMetricsReporterConfig.AUTO_BALANCER_METRICS_REPORTER_LINGER_MS_CONFIG).toString());
        setIfAbsent(producerProps, ProducerConfig.BATCH_SIZE_CONFIG,
                reporterConfig.getInt(AutoBalancerMetricsReporterConfig.AUTO_BALANCER_METRICS_REPORTER_BATCH_SIZE_CONFIG).toString());
        setIfAbsent(producerProps, ProducerConfig.RETRIES_CONFIG, "5");
        setIfAbsent(producerProps, ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.GZIP.toString());
        setIfAbsent(producerProps, ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        setIfAbsent(producerProps, ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, MetricSerde.class.getName());
        setIfAbsent(producerProps, ProducerConfig.ACKS_CONFIG, "all");
        StaticAutoBalancerConfigUtils.addSslConfigs(producerProps, staticAutoBalancerConfig);

        metricsReporterCreateRetries = reporterConfig.getInt(
                AutoBalancerMetricsReporterConfig.AUTO_BALANCER_METRICS_REPORTER_CREATE_RETRIES_CONFIG);

        createAutoBalancerMetricsProducer(producerProps);
        if (producer == null) {
            this.close();
        }

        brokerId = Integer.parseInt((String) configs.get(KRaftConfigs.NODE_ID_CONFIG));
        brokerRack = (String) configs.get(ServerConfigs.BROKER_RACK_CONFIG);
        if (brokerRack == null) {
            brokerRack = "";
        }

        reportingIntervalMs = reporterConfig.getLong(AutoBalancerMetricsReporterConfig.AUTO_BALANCER_METRICS_REPORTER_INTERVAL_MS_CONFIG);

        LOGGER.info("AutoBalancerMetricsReporter configuration finished");
    }
On kafka metric removed

@param metric {@link KafkaMetric}
core/src/main/java/kafka/autobalancer/metricsreporter/AutoBalancerMetricsReporter.java