I have a spring boot application with the following configuration for Kafka consumers:

@EnableKafka
@Configuration
class KafkaConsumerConfig(
    @Value("\${aws.secret-manager.sasl-auth.secret-name}") private val kafkaAuthSecretName: String,
    private val kafkaProperties: KafkaProperties,
    private val awsSecretManagerAdaptor: AwsSecretManagerAdaptor,
    private val applicationContext: ApplicationContext
) {

    private val logger = KotlinLogging.logger { }

    @Bean
    fun kafkaListenerContainerFactory():
        KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> {
        val factory = ConcurrentKafkaListenerContainerFactory<String, String>()
        factory.consumerFactory = consumerFactory()
        factory.containerProperties.isMissingTopicsFatal = false
        factory.setRetryTemplate(retryTemplate())
        factory.setErrorHandler { exception, data ->
            logger.error("Error in process with Exception {} and the record is {}", exception, data)
            SpringApplication.exit(applicationContext)
        }
        return factory
    }

    fun consumerFactory(): ConsumerFactory<String, String> {
        return DefaultKafkaConsumerFactory(consumerConfig())
    }

    fun consumerConfig(): Map<String, Any> {
        val props = kafkaProperties.buildConsumerProperties()
        props[ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java
        props[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = KafkaAvroDeserializer::class.java

        val secretValue = awsSecretManagerAdaptor.getSecretValue(kafkaAuthSecretName)
        val username = getUsername()
        val saslJaasConfig =
            """org.apache.kafka.common.security.scram.ScramLoginModule required username="$username" password="$secretValue";"""
        props[SaslConfigs.SASL_JAAS_CONFIG] = saslJaasConfig

        return props
    }

    private fun getUsername(): String? {
        val secretTags = awsSecretManagerAdaptor.getSecretTags(kafkaAuthSecretName)
        return secretTags.firstOrNull { it.key().equals("username") }?.value()
    }

    private fun retryTemplate(): RetryTemplate {
        val retryTemplate = RetryTemplate()
        retryTemplate.setBackOffPolicy(getFixedBackOffPolicy())
        retryTemplate.setRetryPolicy(getSimpleRetryPolicy())

        return retryTemplate
    }

    private fun getFixedBackOffPolicy(): BackOffPolicy {
        val fixedBackOffPolicy = FixedBackOffPolicy()
        fixedBackOffPolicy.backOffPeriod = 3000
        return fixedBackOffPolicy
    }

    private fun getSimpleRetryPolicy(): SimpleRetryPolicy {
        val simpleRetryPolicy = SimpleRetryPolicy()
        simpleRetryPolicy.maxAttempts = 3
        return simpleRetryPolicy
    }
}

The Kafka server provides SASL Authentication with username and password. As You can see the username and password are fetched using a service called AwsSecretManagerAdaptor from AWS secret manager. The configuration works like charm, however, once the Secret is rotated, and the Kafka consumers are restarted the SASL authentication fails. To solve the issue I am now restarting the SpringBoot Application in order to read the rotated key correctly from AWS Secret Manager.

The solution works but as you can see restarting the application is very ugly and error-prone. I wonder do you have any better suggestions for the improvement?