kafka_python_msk_iam_sasl_authentication_producer.py

1from kafka import KafkaProducer
2from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
3import socket
4
5class MSKTokenProvider():
6    def __init__(self, region):
7        self.region = region
8
9    def token(self):
10        token, _ = MSKAuthTokenProvider.generate_auth_token(self.region)
11        return token
12
13region = 'us-east-1'
14tp = MSKTokenProvider(region)
15
16producer = KafkaProducer(
17    bootstrap_servers='your_broker_address',
18    security_protocol='SASL_SSL',
19    sasl_mechanism='AWS_MSK_IAM',
20    sasl_oauth_token_provider=tp,
21    client_id=socket.gethostname(),
22)
23
24producer.send('your_topic', b'message-from-python-sdk')
25producer.flush()