[HELP] - Kafka Producer IAM Role Authentication Fails After Idle Period with MSK

[Sivakajan-Galaxy]

I need to create a Kafka producer using the confluent_kafka.Producer that works with AWS MSK using IAM role-based authentication. To achieve this, I have developed a working solution as outlined below.

import json
import logging
from confluent_kafka import Producer
from config.settings import KAFKA_BOOTSTRAP_SERVERS, REGION
from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
import socket

logger = logging.getLogger(__name__)


class KafkaProducerService:
    def __init__(self):
        self.producer = Producer(
            {
                "bootstrap.servers": KAFKA_BOOTSTRAP_SERVERS,
                "acks": "all",
                "security.protocol": "SASL_SSL",
                "sasl.mechanism": "OAUTHBEARER",
                "oauth_cb": self.oauth_cb,
                "client.id": socket.gethostname()
            }
        )

    def oauth_cb(self, oauth_config):
        auth_token, exp_ms = MSKAuthTokenProvider.generate_auth_token(REGION)
        return auth_token, exp_ms/1000

    def produce_message(self, topic, message):
        try:
            self.producer.produce(topic=topic, value=json.dumps(message).encode("utf-8"))
            self.producer.flush()
            self.producer.poll(10)
            logger.info(f"Produced message to {topic}: {message}")
        except Exception as e:
            logger.error(f"Failed to send message to {topic}: {str(e)}")
            raise


kafka_producer_service = KafkaProducerService()

The producer is able to establish a healthy connection and successfully send messages to the consumer. However, if the connection remains idle for a few minutes, it starts logging MSK-related connection errors repeatedly.

|1743698633.362|FAIL|api-service-29457134-mr5tm#producer-2| [thrd:sasl_ssl://b-1.*******devmsk.dc43-1948-12.kafka.us-east-1.a]: sasl_ssl://b-1.********devmsk.dc256562362.kafka.us-east-1.amazonaws.com:9098/1: SASL authentication error: [20234d-243-49fb-92e8-4]2324234: Access denied (after 235ms in state AUTH_REQ)

The error logs keep appearing repeatedly while the connection is idle. However, as soon as a new message is sent, it reaches the consumer successfully and the error logs stop. Once the connection becomes idle again, the same errors start appearing after a short timeout.

Can anyone help me to solve this, and able to give a best solution for the case.