spring-projects/spring-kafka

I have changed my implementation to use spring-kafka and still see the same issue of kafka metrics not being exposed

@EnableKafka
@Configuration
class KafkaConsumerConfig {
    private static final Logger logger = LoggerFactory.getLogger(KafkaConsumerConfig.class);

    @Autowired
    private GlobalConfig config;

    @Bean
    public ConsumerFactory<byte[], RheosEvent> salesTopicConsumerFactory() {
        Map<String, Object> props = commonConfig();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "XXX");
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, config.getConsumerName() + ".zzz");
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean(name="salesKafkaListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<byte[], RheosEvent>> salesKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<byte[], RheosEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConcurrency(1);
        factory.setBatchListener(true);
        factory.setConsumerFactory(salesTopicConsumerFactory());
        factory.getContainerProperties().setConsumerRebalanceListener(new CustomRebalanceListener());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

    @Bean
    public ConsumerFactory<byte[], RheosEvent> behaviorTopicConsumerFactory() {
        Map<String, Object> props = commonConfig();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "YYY");
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, config.getConsumerName());
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean(name="behaviorKafkaListenerContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<byte[], RheosEvent>> behaviorKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<byte[], RheosEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConcurrency(1);
        factory.setBatchListener(true);
        factory.setConsumerFactory(behaviorTopicConsumerFactory());
        factory.getContainerProperties().setConsumerRebalanceListener(new CustomRebalanceListener());
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }

    private Map<String, Object> commonConfig() {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.GROUP_ID_CONFIG, config.getGroupId());
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, config.getKeyDeserializer());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, config.getValueDeserializer());
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, config.getEnableAutoCommit());
        props.put(ConsumerConfig.RECEIVE_BUFFER_CONFIG, config.getReceiveBufferBytes());
        props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, config.getMaxPartitionFetchBytes());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, config.getMaxPollRecords());
        props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, RoundRobinAssignor.class.getName());
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 1000);
        return props;
    }

    static class CustomRebalanceListener implements ConsumerAwareRebalanceListener{
        @Override
        public void onPartitionsRevokedBeforeCommit(Consumer<?, ?> consumer, Collection<TopicPartition> partitions) {
            logger.info("PARTITIONS revoked: " + partitions);
            consumer.commitAsync();
        }

        @Override
        public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            logger.info("PARTITIONS assigned: " + partitions);
        }
    }
}

Please suggest if I am missing any configuration