aws_msk_iam_sasl_signer_kafka_python_producer_auth.py

1import pandas as pd
2from kafka import KafkaProducer
3from aws_msk_iam_sasl_signer import MSKAuthTokenProvider
4
5# Define a class to provide the MSK IAM authentication token
6class MSKTokenProvider():
7    def __init__(self, region):
8        self.region = region
9
10    def token(self):
11        token, _ = MSKAuthTokenProvider.generate_auth_token(self.region)
12        return token
13
14# Replace with your MSK cluster details
15region = 'us-east-1'
16tp = MSKTokenProvider(region)
17broker_address = 'your-msk-broker-address:9098'
18
19# Initialize the KafkaProducer with IAM authentication
20producer = KafkaProducer(
21    bootstrap_servers=broker_address,
22    security_protocol='SASL_SSL',
23    sasl_mechanism='AWS_MSK_IAM',
24    sasl_oauth_token_provider=tp,
25    client_id='client1',
26)
27
28# Send a test message
29producer.send('your-topic-name', b'Hello MSK with IAM!')
30producer.flush()
31print("Message sent successfully")